Aggiunge registry nodi stream e provisioning Hetzner/lab per lo scale-out.
Prepara l'architettura multi-nodo (MediaMTX+ffmpeg) con assignment URL per sessione, admin di provision/drain e provider Cloud/DNS astratti verso mltv-stream.net. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
# frozen_string_literal: true
|
||||
|
||||
require "faraday"
|
||||
|
||||
module Streams
|
||||
module CloudProviders
|
||||
# Clone/start/stop di VM template su Proxmox VE (API token).
|
||||
#
|
||||
# ENV richiesti:
|
||||
# PROXMOX_API_URL, PROXMOX_TOKEN_ID, PROXMOX_TOKEN_SECRET,
|
||||
# PROXMOX_NODE, PROXMOX_TEMPLATE_VMID
|
||||
class ProxmoxLab < Base
|
||||
def initialize(
|
||||
api_url: ENV.fetch("PROXMOX_API_URL"),
|
||||
token_id: ENV.fetch("PROXMOX_TOKEN_ID"),
|
||||
token_secret: ENV.fetch("PROXMOX_TOKEN_SECRET"),
|
||||
node: ENV.fetch("PROXMOX_NODE"),
|
||||
template_vmid: ENV.fetch("PROXMOX_TEMPLATE_VMID"),
|
||||
verify_ssl: ENV.fetch("PROXMOX_VERIFY_SSL", "false") == "true"
|
||||
)
|
||||
@api_url = api_url.to_s.chomp("/")
|
||||
@token_id = token_id
|
||||
@token_secret = token_secret
|
||||
@node = node
|
||||
@template_vmid = template_vmid.to_i
|
||||
@verify_ssl = verify_ssl
|
||||
end
|
||||
|
||||
def create_node(name:, labels: {})
|
||||
newid = next_vmid
|
||||
post("/nodes/#{@node}/qemu/#{@template_vmid}/clone", {
|
||||
newid: newid,
|
||||
name: name,
|
||||
full: 1,
|
||||
target: @node
|
||||
})
|
||||
post("/nodes/#{@node}/qemu/#{newid}/status/start", {})
|
||||
wait_until_running(newid.to_s)
|
||||
ip = public_ip(newid.to_s)
|
||||
Instance.new(
|
||||
id: newid.to_s,
|
||||
name: name,
|
||||
public_ip: ip,
|
||||
private_ip: ip,
|
||||
status: "running",
|
||||
raw: { node: @node, vmid: newid, labels: labels }
|
||||
)
|
||||
end
|
||||
|
||||
def destroy_node(instance_id)
|
||||
vmid = instance_id.to_i
|
||||
begin
|
||||
post("/nodes/#{@node}/qemu/#{vmid}/status/stop", { timeout: 30 })
|
||||
rescue Error
|
||||
# già spenta
|
||||
end
|
||||
sleep 2
|
||||
delete("/nodes/#{@node}/qemu/#{vmid}", { purge: 1 })
|
||||
true
|
||||
end
|
||||
|
||||
def list_nodes(labels: {})
|
||||
items = get("/nodes/#{@node}/qemu")
|
||||
Array(items).filter_map do |row|
|
||||
name = row["name"].to_s
|
||||
next unless name.start_with?("mltv-stream-") || name.start_with?("ingest-")
|
||||
|
||||
Instance.new(
|
||||
id: row["vmid"].to_s,
|
||||
name: name,
|
||||
public_ip: nil,
|
||||
private_ip: nil,
|
||||
status: row["status"],
|
||||
raw: row
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
def wait_until_running(instance_id, timeout: 180)
|
||||
deadline = Time.now + timeout
|
||||
loop do
|
||||
status = get("/nodes/#{@node}/qemu/#{instance_id}/status/current")
|
||||
return Instance.new(id: instance_id.to_s, status: "running", raw: status) if status["status"] == "running"
|
||||
raise Error, "Timeout attesa VM #{instance_id}" if Time.now >= deadline
|
||||
|
||||
sleep 3
|
||||
end
|
||||
end
|
||||
|
||||
def public_ip(instance_id)
|
||||
agent = get("/nodes/#{@node}/qemu/#{instance_id}/agent/network-get-interfaces")
|
||||
interfaces = agent.is_a?(Hash) ? agent["result"] : nil
|
||||
Array(interfaces).each do |iface|
|
||||
Array(iface["ip-addresses"]).each do |addr|
|
||||
ip = addr["ip-address"].to_s
|
||||
next if ip.blank? || ip.start_with?("127.") || ip.include?(":")
|
||||
|
||||
return ip
|
||||
end
|
||||
end
|
||||
ENV["STREAM_LAB_FALLBACK_IP"].presence || "127.0.0.1"
|
||||
rescue Error
|
||||
ENV["STREAM_LAB_FALLBACK_IP"].presence || "127.0.0.1"
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def next_vmid
|
||||
used = Array(get("/cluster/resources", type: "vm")).map { |r| r["vmid"].to_i }
|
||||
candidate = ENV.fetch("PROXMOX_VMID_START", "9100").to_i
|
||||
candidate += 1 while used.include?(candidate)
|
||||
candidate
|
||||
end
|
||||
|
||||
def conn
|
||||
@conn ||= Faraday.new(url: "#{@api_url}/api2/json") do |f|
|
||||
f.request :url_encoded
|
||||
f.response :json, content_type: /\bjson$/
|
||||
f.adapter Faraday.default_adapter
|
||||
f.ssl[:verify] = @verify_ssl
|
||||
end
|
||||
end
|
||||
|
||||
def auth_headers
|
||||
{ "Authorization" => "PVEAPIToken=#{@token_id}=#{@token_secret}" }
|
||||
end
|
||||
|
||||
def get(path, params = {})
|
||||
response = conn.get(path) do |req|
|
||||
req.headers.update(auth_headers)
|
||||
req.params.update(params)
|
||||
end
|
||||
unwrap!(response)
|
||||
end
|
||||
|
||||
def post(path, body = {})
|
||||
response = conn.post(path) do |req|
|
||||
req.headers.update(auth_headers)
|
||||
req.body = body
|
||||
end
|
||||
unwrap!(response)
|
||||
end
|
||||
|
||||
def delete(path, params = {})
|
||||
response = conn.delete(path) do |req|
|
||||
req.headers.update(auth_headers)
|
||||
req.params.update(params)
|
||||
end
|
||||
unwrap!(response)
|
||||
end
|
||||
|
||||
def unwrap!(response)
|
||||
unless response.success?
|
||||
raise Error, "Proxmox API #{response.status}: #{response.body.inspect}"
|
||||
end
|
||||
|
||||
body = response.body
|
||||
body.is_a?(Hash) && body.key?("data") ? body["data"] : body
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
Reference in New Issue
Block a user