diff --git a/backend/app/models/stream_node.rb b/backend/app/models/stream_node.rb index 02954d3..73bfcbf 100644 --- a/backend/app/models/stream_node.rb +++ b/backend/app/models/stream_node.rb @@ -35,4 +35,20 @@ class StreamNode < ApplicationRecord def allocatable? status == "ready" && free_slots.positive? end + + # Agent ffmpeg sul CPX (cloud-init :9100). Home Sidekiq lo chiama senza worker Docker per nodo. + def relay_agent_url + return unless role == "cloud" + + ip = (metadata || {})["public_ip"].presence + if ip.blank? + host = URI.parse(api_base_url.to_s).host + ip = host if host.present? && host != "localhost" + end + return if ip.blank? || ip == "127.0.0.1" + + "http://#{ip}:#{ENV.fetch("STREAM_NODE_AGENT_PORT", "9100")}" + rescue URI::InvalidURIError + nil + end end diff --git a/backend/app/services/streams/youtube_relay.rb b/backend/app/services/streams/youtube_relay.rb index 51e4c6e..5f8d8f4 100644 --- a/backend/app/services/streams/youtube_relay.rb +++ b/backend/app/services/streams/youtube_relay.rb @@ -35,7 +35,26 @@ module Streams assigned_node_slug(session) == local_node_slug end + # Home può avviare ffmpeg sull'agent del CPX (niente worker Docker per nodo). + def can_process?(session) + return false unless worker? + return true if node_matches?(session) + + cloud_home_dispatch?(session) + end + + def cloud_home_dispatch?(session) + local_node_slug == NodeRegistry::HOME_SLUG && + session.stream_node&.role == "cloud" && + session.stream_node.relay_agent_url.present? + end + def queue_for(session) + node = session.stream_node + if node&.role == "cloud" && node.relay_agent_url.present? + return "#{QUEUE_PREFIX}_cloud" + end + "#{QUEUE_PREFIX}_#{assigned_node_slug(session)}" end @@ -50,14 +69,14 @@ module Streams end def start(session) - return enqueue_ensure!(session) unless worker? && node_matches?(session) + return enqueue_ensure!(session) unless can_process?(session) start_on_worker!(session) end # Non cancella owner/pid qui: solo il worker owner deve killare ffmpeg. def stop(session) - if worker? && owner_is_local?(session.id) && node_matches?(session) + if worker? && owner_is_local?(session.id) && can_process?(session) stop_on_worker!(session) else enqueue_stop!(session) @@ -83,7 +102,7 @@ module Streams return if session.terminal? return if session.stream_key.blank? - if worker? && node_matches?(session) + if can_process?(session) ensure_on_worker!(session) else enqueue_ensure!(session) @@ -92,7 +111,7 @@ module Streams def ensure_on_worker!(session) return :not_worker unless worker? - unless node_matches?(session) + unless can_process?(session) enqueue_ensure!(session, wait: 2.seconds) Rails.logger.info( "[YoutubeRelay] wrong_node local=#{local_node_slug} assigned=#{assigned_node_slug(session)} session=#{session.id}" @@ -115,7 +134,7 @@ module Streams return :already_running end - if at_capacity? + if at_capacity_for?(session) enqueue_ensure!(session, wait: 5.seconds) Rails.logger.info("[YoutubeRelay] at capacity worker=#{worker_id} node=#{local_node_slug} session=#{session.id} requeue") return :at_capacity @@ -129,6 +148,7 @@ module Streams :started rescue Error => e Rails.logger.warn("[YoutubeRelay] ensure_on_worker session=#{session.id}: #{e.message}") + enqueue_ensure!(session, wait: 5.seconds) unless session.terminal? :error end @@ -161,6 +181,13 @@ module Streams local_owned_count >= max_concurrent end + # I relay sull'agent CPX non consumano ffmpeg locale: non applicare il cap del worker home. + def at_capacity_for?(session) + return false if cloud_home_dispatch?(session) + + at_capacity? + end + def owner_is_local?(session_id) owner = redis.get(format(OWNER_KEY, session_id)) owner.blank? || owner == worker_id @@ -188,7 +215,7 @@ module Streams intake_source = mediamtx_intake_source(session) output = "rtmps://a.rtmps.youtube.com/live2/#{session.stream_key}" - pid = if relay_agent_enabled? + pid = if agent_dispatch?(session) start_via_agent!(session) else spawn_ffmpeg!(youtube_ffmpeg_args(intake_source, output), log_path) @@ -197,7 +224,7 @@ module Streams claim_ownership!(session.id) Rails.logger.info( "[YoutubeRelay] started pid=#{pid} session=#{session.id} intake=#{intake_source.join(':')} " \ - "worker=#{worker_id} agent=#{relay_agent_url || '-'}" + "worker=#{worker_id} agent=#{agent_base_url(session) || '-'}" ) schedule_youtube_activate(session) pid @@ -205,16 +232,18 @@ module Streams raise Error, "ffmpeg non disponibile: #{e.message}" end - def relay_agent_url - ENV["STREAM_NODE_RELAY_AGENT_URL"].presence + def agent_base_url(session) + return if session.blank? + + ENV["STREAM_NODE_RELAY_AGENT_URL"].presence || session.stream_node&.relay_agent_url end - def relay_agent_enabled? - relay_agent_url.present? + def agent_dispatch?(session) + agent_base_url(session).present? end def relay_agent_secret - ENV["STREAM_NODE_AGENT_SECRET"].presence || ENV["MEDIAMTX_WEBHOOK_SECRET"].presence || "" + ENV["STREAM_NODE_AGENT_SECRET"].presence || "mediamtx_webhook_dev_secret" end def spawn_ffmpeg!(args, log_path) @@ -229,19 +258,23 @@ module Streams "path" => session.mediamtx_path_name, "rtmps" => "rtmps://a.rtmps.youtube.com/live2/#{session.stream_key}" } - res = agent_request(Net::HTTP::Post, "/relays", payload) + res = agent_request(Net::HTTP::Post, "/relays", payload, session: session) pid = res["pid"].to_i raise Error, "relay agent spawn failed: #{res.inspect}" if pid <= 0 File.write( log_file(session), - "agent=#{relay_agent_url} pid=#{pid} path=#{session.mediamtx_path_name}\n" + "agent=#{agent_base_url(session)} pid=#{pid} path=#{session.mediamtx_path_name}\n" ) pid end - def agent_request(http_class, path, payload = nil) - uri = URI.parse("#{relay_agent_url.chomp('/')}#{path}") + def agent_request(http_class, path, payload = nil, session: nil, session_id: nil) + session ||= StreamSession.find_by(id: session_id) if session_id.present? + base = agent_base_url(session) + raise Error, "relay agent URL mancante" if base.blank? + + uri = URI.parse("#{base.chomp('/')}#{path}") http = Net::HTTP.new(uri.host, uri.port) http.open_timeout = 5 http.read_timeout = 10 @@ -304,7 +337,7 @@ module Streams # Sul nodo assegnato ffmpeg legge MediaMTX in loopback (o hostname Docker locale). def intake_bases(session) - if relay_agent_enabled? && node_matches?(session) + if agent_dispatch?(session) && (node_matches?(session) || cloud_home_dispatch?(session)) return ["rtmp://127.0.0.1:1935", "http://127.0.0.1:8888"] end @@ -376,7 +409,8 @@ module Streams end def process_alive?(pid, session_id: nil) - return agent_session_running?(session_id) if relay_agent_enabled? && session_id.present? + session = session_id.present? ? StreamSession.find_by(id: session_id) : nil + return agent_session_running?(session) if session && agent_dispatch?(session) stat = File.read("/proc/#{pid.to_i}/stat") return false if stat.split[2] == "Z" @@ -387,16 +421,17 @@ module Streams false end - def agent_session_running?(session_id) - res = agent_request(Net::HTTP::Get, "/relays/#{session_id}") + def agent_session_running?(session) + res = agent_request(Net::HTTP::Get, "/relays/#{session.id}", session: session) res["running"] == true rescue Error false end def terminate_pid(pid, session_id: nil) - if relay_agent_enabled? && session_id.present? - agent_request(Net::HTTP::Delete, "/relays/#{session_id}") + session = session_id.present? ? StreamSession.find_by(id: session_id) : nil + if session && agent_dispatch?(session) + agent_request(Net::HTTP::Delete, "/relays/#{session_id}", session: session) return end @@ -405,7 +440,7 @@ module Streams rescue Errno::ESRCH nil else - return if relay_agent_enabled? + return if session && agent_dispatch?(session) begin Process.kill("KILL", -pid.to_i) diff --git a/backend/config/sidekiq.yml b/backend/config/sidekiq.yml index 186e04d..cbb31dd 100644 --- a/backend/config/sidekiq.yml +++ b/backend/config/sidekiq.yml @@ -2,5 +2,6 @@ :queues: - critical - youtube_relay_home + - youtube_relay_cloud - youtube_relay - default diff --git a/backend/spec/services/streams/youtube_relay_spec.rb b/backend/spec/services/streams/youtube_relay_spec.rb index 6e40d9a..ef44085 100644 --- a/backend/spec/services/streams/youtube_relay_spec.rb +++ b/backend/spec/services/streams/youtube_relay_spec.rb @@ -33,7 +33,15 @@ RSpec.describe Streams::YoutubeRelay do end let(:session_id) { SecureRandom.uuid } - let(:lab_node) { instance_double(StreamNode, slug: "ingest-lab-01") } + let(:lab_node) { instance_double(StreamNode, slug: "ingest-lab-01", role: "lab", relay_agent_url: nil) } + let(:cloud_node) do + instance_double( + StreamNode, + slug: "ingest-01", + role: "cloud", + relay_agent_url: "http://203.0.113.10:9100" + ) + end def session_double(stream_node: nil) instance_double( @@ -54,10 +62,14 @@ RSpec.describe Streams::YoutubeRelay do expect(described_class.queue_for(session_double)).to eq("youtube_relay_home") end - it "routes overflow sessions to the node slug queue" do + it "routes overflow lab sessions to the node slug queue" do expect(described_class.queue_for(session_double(stream_node: lab_node))).to eq("youtube_relay_ingest-lab-01") end + it "routes cloud overflow to youtube_relay_cloud for the home dispatcher" do + expect(described_class.queue_for(session_double(stream_node: cloud_node))).to eq("youtube_relay_cloud") + end + it "requeues ensure to the assigned node when the local worker does not match" do with_env("STREAM_NODE_SLUG" => "home", "YOUTUBE_RELAY_WORKER" => "1") do allow(described_class).to receive(:worker?).and_return(true) @@ -67,6 +79,13 @@ RSpec.describe Streams::YoutubeRelay do expect(described_class.ensure_on_worker!(session_double(stream_node: lab_node))).to eq(:wrong_host) end end + + it "lets the home worker process cloud overflow instead of requeueing to a per-node queue" do + with_env("STREAM_NODE_SLUG" => "home", "YOUTUBE_RELAY_WORKER" => "1") do + expect(described_class.can_process?(session_double(stream_node: cloud_node))).to be true + expect(described_class.cloud_home_dispatch?(session_double(stream_node: cloud_node))).to be true + end + end end describe "multi-host sticky stop/capacity" do @@ -125,7 +144,7 @@ RSpec.describe Streams::YoutubeRelay do let(:session) do instance_double( StreamSession, - stream_node: instance_double(StreamNode, slug: "ingest-lab-01"), + stream_node: instance_double(StreamNode, slug: "ingest-lab-01", role: "lab", relay_agent_url: nil), mediamtx_path_name: "live/match_abc", mediamtx_internal_rtmp_url: "rtmp://10.0.0.9:1935", mediamtx_internal_hls_url: "http://10.0.0.9:8888" @@ -179,7 +198,8 @@ RSpec.describe Streams::YoutubeRelay do expect(described_class).to receive(:agent_request).with( Net::HTTP::Post, "/relays", - hash_including("session_id" => "sess-1", "path" => "live/match_abc") + hash_including("session_id" => "sess-1", "path" => "live/match_abc"), + session: session ).and_return("pid" => 4242) pid = with_env("STREAM_NODE_RELAY_AGENT_URL" => "http://203.0.113.10:9100") do @@ -189,3 +209,27 @@ RSpec.describe Streams::YoutubeRelay do end end end + +RSpec.describe StreamNode do + describe "#relay_agent_url" do + it "uses metadata public_ip on cloud nodes" do + node = described_class.new( + slug: "ingest-01", + hostname: "ingest-01.mltv-stream.net", + role: "cloud", + status: "ready", + provider: "hetzner", + rtmp_base_url: "rtmp://x:1935", + hls_base_url: "https://x/hls", + api_base_url: "http://10.0.0.9:9997", + metadata: { "public_ip" => "203.0.113.9" } + ) + expect(node.relay_agent_url).to eq("http://203.0.113.9:9100") + end + + it "is nil for lab nodes" do + node = described_class.new(role: "lab", api_base_url: "http://10.0.0.9:9997", metadata: { "public_ip" => "1.2.3.4" }) + expect(node.relay_agent_url).to be_nil + end + end +end diff --git a/docs/infrastructure/STREAMING_AUTOSCALE.md b/docs/infrastructure/STREAMING_AUTOSCALE.md index afd45c0..d44791f 100644 --- a/docs/infrastructure/STREAMING_AUTOSCALE.md +++ b/docs/infrastructure/STREAMING_AUTOSCALE.md @@ -255,12 +255,12 @@ L’overflow deve spostare **ffmpeg**, non solo l’ingest RTMP. In lab `local_l ### Fase 2 — dettagli implementati -- Coda Sidekiq dedicata `youtube_relay_` (home: `youtube_relay_home`; overflow: `youtube_relay_ingest-01`) -- `YoutubeRelayEnsureJob` / `YoutubeRelayStopJob` sulla coda del **nodo assegnato** +- Coda Sidekiq `youtube_relay_home` (ffmpeg locale) e `youtube_relay_cloud` (home chiama l’agent `:9100` sul CPX). Lab overflow resta `youtube_relay_` +- `YoutubeRelayEnsureJob` / `YoutubeRelayStopJob` sulla coda del **nodo assegnato** (cloud → `youtube_relay_cloud`) - Stop sticky: non cancella owner da Rails; lo stop gira sull’owner o requeue (`:wrong_host`) -- Cap per worker: `RELAY_MAX_CONCURRENT` (default 4) + set Redis `youtube_relay:owned:HOSTNAME` +- Cap per worker: `RELAY_MAX_CONCURRENT` (default 4) + set Redis `youtube_relay:owned:HOSTNAME` (non si applica al dispatch agent dal home) - Ensure a capacità piena → requeue 5s sulla stessa coda nodo -- Worker home ascolta `youtube_relay_home` (+ `youtube_relay` legacy). Worker overflow: solo la propria coda, **niente** `default`/`critical` +- Worker home ascolta `youtube_relay_home`, `youtube_relay_cloud` (+ `youtube_relay` legacy). Un CPX nuovo non richiede `prod_relay_worker.sh` - Intake locale: `STREAM_NODE_LOCAL_RTMP_URL` (loopback sul CPX, `mediamtx` in Docker lab) ### Fase 3 — dettagli implementati diff --git a/infra/.env.production.example b/infra/.env.production.example index 0309bd7..e995fa0 100644 --- a/infra/.env.production.example +++ b/infra/.env.production.example @@ -125,6 +125,8 @@ STREAM_NODE_ENV=prod STREAM_CLOUD_PROVIDER=local_lab RELAY_MAX_CONCURRENT=4 YOUTUBE_RELAY_WORKER=1 +# Deve coincidere con cloud-init (default mediamtx_webhook_dev_secret) — home chiama l'agent :9100. +STREAM_NODE_AGENT_SECRET=mediamtx_webhook_dev_secret # Autoscaler (kill-switch: 0 finché WireGuard/smoke Cloud non sono OK — NON abilitare in prod senza lab) STREAM_AUTOSCALE_ENABLED=0 diff --git a/infra/docker-compose.prod.yml b/infra/docker-compose.prod.yml index 6ff2cf8..69d81dd 100644 --- a/infra/docker-compose.prod.yml +++ b/infra/docker-compose.prod.yml @@ -53,6 +53,8 @@ services: dockerfile: Dockerfile restart: unless-stopped command: bundle exec rails server -b 0.0.0.0 -p 3000 -e production + env_file: + - .env environment: RAILS_ENV: production RAILS_MAX_THREADS: ${RAILS_MAX_THREADS:-5} @@ -70,6 +72,33 @@ services: YOUTUBE_CLIENT_SECRET: ${YOUTUBE_CLIENT_SECRET:-} YOUTUBE_REDIRECT_URI: ${YOUTUBE_REDIRECT_URI:-} YOUTUBE_PLATFORM_REFRESH_TOKEN: ${YOUTUBE_PLATFORM_REFRESH_TOKEN:-} + STREAM_NODE_AGENT_SECRET: ${STREAM_NODE_AGENT_SECRET:-mediamtx_webhook_dev_secret} + STREAM_NODE_HOME_MAX_PUBLISHERS: ${STREAM_NODE_HOME_MAX_PUBLISHERS:-6} + STREAM_CLOUD_PROVIDER: ${STREAM_CLOUD_PROVIDER:-local_lab} + STREAM_DNS_PROVIDER: ${STREAM_DNS_PROVIDER:-hetzner} + STREAM_AUTOSCALE_ENABLED: ${STREAM_AUTOSCALE_ENABLED:-0} + STREAM_AUTOSCALE_KIND: ${STREAM_AUTOSCALE_KIND:-lab} + STREAM_AUTOSCALE_ALLOW_CLOUD: ${STREAM_AUTOSCALE_ALLOW_CLOUD:-0} + STREAM_AUTOSCALE_SOFT_FREE_SLOTS: ${STREAM_AUTOSCALE_SOFT_FREE_SLOTS:-2} + STREAM_AUTOSCALE_WARM_SPARE: ${STREAM_AUTOSCALE_WARM_SPARE:-1} + STREAM_AUTOSCALE_MAX_NODES: ${STREAM_AUTOSCALE_MAX_NODES:-3} + STREAM_AUTOSCALE_IDLE_MINUTES: ${STREAM_AUTOSCALE_IDLE_MINUTES:-30} + STREAM_AUTOSCALE_INTERVAL_SECS: ${STREAM_AUTOSCALE_INTERVAL_SECS:-60} + STREAM_AUTOSCALE_NODE_EUR_PER_HOUR: ${STREAM_AUTOSCALE_NODE_EUR_PER_HOUR:-0.015} + STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR: ${STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR:-40} + STREAM_CLOUD_PUBLIC_CONTROL: ${STREAM_CLOUD_PUBLIC_CONTROL:-1} + STREAM_NODE_ENV: ${STREAM_NODE_ENV:-prod} + STREAM_DNS_ZONE: ${STREAM_DNS_ZONE:-mltv-stream.net} + STREAM_DNS_TTL: ${STREAM_DNS_TTL:-60} + STREAM_CLOUD_DNS_SUFFIX: ${STREAM_CLOUD_DNS_SUFFIX:-mltv-stream.net} + STREAM_CLOUD_MAX_PUBLISHERS: ${STREAM_CLOUD_MAX_PUBLISHERS:-4} + HCLOUD_TOKEN: ${HCLOUD_TOKEN:-} + HCLOUD_LOCATION: ${HCLOUD_LOCATION:-nbg1} + HCLOUD_SERVER_TYPE: ${HCLOUD_SERVER_TYPE:-cpx12} + HCLOUD_IMAGE: ${HCLOUD_IMAGE:-debian-12} + HCLOUD_SSH_KEY: ${HCLOUD_SSH_KEY:-matchlivetv-stream-hetzner} + HCLOUD_NETWORK_ID: ${HCLOUD_NETWORK_ID:-} + HCLOUD_USER_DATA_FILE: /opt/matchlivetv/infra/stream-node/cloud-init.yaml HLS_PUBLIC_URL: ${HLS_PUBLIC_URL:-http://localhost:8888} APP_PUBLIC_URL: ${APP_PUBLIC_URL:-http://localhost:3000} PRIVACY_CONTACT_EMAIL: ${PRIVACY_CONTACT_EMAIL:-privacy@matchlivetv.it} @@ -166,6 +195,8 @@ services: restart: unless-stopped hostname: home command: bundle exec sidekiq -C config/sidekiq.yml -e production + env_file: + - .env environment: RAILS_ENV: production RAILS_MAX_THREADS: ${RAILS_MAX_THREADS:-5} @@ -182,6 +213,33 @@ services: YOUTUBE_PLATFORM_REFRESH_TOKEN: ${YOUTUBE_PLATFORM_REFRESH_TOKEN:-} YOUTUBE_RELAY_WORKER: "1" STREAM_NODE_SLUG: home + STREAM_NODE_HOME_MAX_PUBLISHERS: ${STREAM_NODE_HOME_MAX_PUBLISHERS:-6} + STREAM_CLOUD_PROVIDER: ${STREAM_CLOUD_PROVIDER:-local_lab} + STREAM_DNS_PROVIDER: ${STREAM_DNS_PROVIDER:-hetzner} + STREAM_AUTOSCALE_ENABLED: ${STREAM_AUTOSCALE_ENABLED:-0} + STREAM_AUTOSCALE_KIND: ${STREAM_AUTOSCALE_KIND:-lab} + STREAM_AUTOSCALE_ALLOW_CLOUD: ${STREAM_AUTOSCALE_ALLOW_CLOUD:-0} + STREAM_AUTOSCALE_SOFT_FREE_SLOTS: ${STREAM_AUTOSCALE_SOFT_FREE_SLOTS:-2} + STREAM_AUTOSCALE_WARM_SPARE: ${STREAM_AUTOSCALE_WARM_SPARE:-1} + STREAM_AUTOSCALE_MAX_NODES: ${STREAM_AUTOSCALE_MAX_NODES:-3} + STREAM_AUTOSCALE_IDLE_MINUTES: ${STREAM_AUTOSCALE_IDLE_MINUTES:-30} + STREAM_AUTOSCALE_INTERVAL_SECS: ${STREAM_AUTOSCALE_INTERVAL_SECS:-60} + STREAM_AUTOSCALE_NODE_EUR_PER_HOUR: ${STREAM_AUTOSCALE_NODE_EUR_PER_HOUR:-0.015} + STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR: ${STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR:-40} + STREAM_CLOUD_PUBLIC_CONTROL: ${STREAM_CLOUD_PUBLIC_CONTROL:-1} + STREAM_NODE_ENV: ${STREAM_NODE_ENV:-prod} + STREAM_DNS_ZONE: ${STREAM_DNS_ZONE:-mltv-stream.net} + STREAM_DNS_TTL: ${STREAM_DNS_TTL:-60} + STREAM_CLOUD_DNS_SUFFIX: ${STREAM_CLOUD_DNS_SUFFIX:-mltv-stream.net} + STREAM_CLOUD_MAX_PUBLISHERS: ${STREAM_CLOUD_MAX_PUBLISHERS:-4} + HCLOUD_TOKEN: ${HCLOUD_TOKEN:-} + HCLOUD_LOCATION: ${HCLOUD_LOCATION:-nbg1} + HCLOUD_SERVER_TYPE: ${HCLOUD_SERVER_TYPE:-cpx12} + HCLOUD_IMAGE: ${HCLOUD_IMAGE:-debian-12} + HCLOUD_SSH_KEY: ${HCLOUD_SSH_KEY:-matchlivetv-stream-hetzner} + HCLOUD_NETWORK_ID: ${HCLOUD_NETWORK_ID:-} + HCLOUD_USER_DATA_FILE: /opt/matchlivetv/infra/stream-node/cloud-init.yaml + STREAM_NODE_AGENT_SECRET: ${STREAM_NODE_AGENT_SECRET:-mediamtx_webhook_dev_secret} STREAM_NODE_LOCAL_RTMP_URL: rtmp://mediamtx:1935 STREAM_NODE_LOCAL_HLS_URL: http://mediamtx:8888 RAILS_INTERNAL_URL: http://rails:3000 diff --git a/infra/stream-node/cloud-init.yaml b/infra/stream-node/cloud-init.yaml index 14b4656..5666e43 100644 --- a/infra/stream-node/cloud-init.yaml +++ b/infra/stream-node/cloud-init.yaml @@ -56,6 +56,210 @@ write_files: STREAM_NODE_AGENT_LISTEN=0.0.0.0:9100 STREAM_NODE_LOCAL_HLS_URL=http://127.0.0.1:8888 STREAM_NODE_RELAY_LOG_DIR=/var/log + - path: /opt/stream-node/relay-agent.py + permissions: "0755" + content: | + #!/usr/bin/env python3 + """Avvia/ferma ffmpeg sul nodo stream (loopback MediaMTX → YouTube). + Il control plane chiama questo agent all'avvio della diretta; ffmpeg resta sul CPX. + """ + from __future__ import annotations + + import json + import os + import signal + import subprocess + import threading + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + from urllib.parse import urlparse + + SECRET = os.environ.get("STREAM_NODE_AGENT_SECRET", "") + LISTEN = os.environ.get("STREAM_NODE_AGENT_LISTEN", "0.0.0.0:9100") + HLS_BASE = os.environ.get("STREAM_NODE_LOCAL_HLS_URL", "http://127.0.0.1:8888").rstrip("/") + LOG_DIR = os.environ.get("STREAM_NODE_RELAY_LOG_DIR", "/var/log") + + _lock = threading.Lock() + # session_id -> {"pid": int, "path": str, "proc": Popen} + _relays: dict[str, dict] = {} + + + def _authorized(handler: BaseHTTPRequestHandler) -> bool: + if not SECRET: + return True + hdr = handler.headers.get("Authorization", "") + return hdr == f"Bearer {SECRET}" + + + def _ffmpeg_cmd(path: str, rtmps: str) -> list[str]: + return [ + "ffmpeg", + "-nostdin", + "-hide_banner", + "-loglevel", + "warning", + "-fflags", + "+genpts+discardcorrupt", + "-reconnect", + "1", + "-reconnect_streamed", + "1", + "-reconnect_on_network_error", + "1", + "-reconnect_delay_max", + "2", + "-rw_timeout", + "15000000", + "-live_start_index", + "-1", + "-i", + f"{HLS_BASE}/{path}/index.m3u8", + "-c:v", + "copy", + "-c:a", + "copy", + "-bsf:a", + "aac_adtstoasc", + "-f", + "flv", + rtmps, + ] + + + def _alive(pid: int) -> bool: + try: + os.kill(pid, 0) + return True + except OSError: + return False + + + def _stop_session(session_id: str) -> None: + info = _relays.pop(session_id, None) + if not info: + return + proc = info.get("proc") + pid = int(info.get("pid") or 0) + if proc and proc.poll() is None: + try: + os.killpg(proc.pid, signal.SIGTERM) + except OSError: + pass + try: + proc.wait(timeout=2) + except Exception: + try: + os.killpg(proc.pid, signal.SIGKILL) + except OSError: + pass + elif pid: + try: + os.kill(pid, signal.SIGTERM) + except OSError: + pass + + + def _start_session(session_id: str, path: str, rtmps: str) -> int: + existing = _relays.get(session_id) + if existing: + proc = existing.get("proc") + pid = int(existing.get("pid") or 0) + if proc is not None and proc.poll() is None and existing.get("path") == path: + return pid + _stop_session(session_id) + os.makedirs(LOG_DIR, exist_ok=True) + safe = path.replace("/", "_") + log_path = os.path.join(LOG_DIR, f"youtube-relay-{safe}.log") + log = open(log_path, "ab", buffering=0) + proc = subprocess.Popen( + _ffmpeg_cmd(path, rtmps), + stdout=log, + stderr=log, + start_new_session=True, + ) + _relays[session_id] = {"pid": proc.pid, "path": path, "proc": proc} + return proc.pid + + + class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt: str, *args) -> None: + print(f"[relay-agent] {self.address_string()} {fmt % args}") + + def _json(self, code: int, payload: dict) -> None: + body = json.dumps(payload).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + parsed = urlparse(self.path) + if parsed.path in ("/health", "/"): + self._json(200, {"ok": True, "relays": len(_relays)}) + return + if not _authorized(self): + self._json(401, {"error": "unauthorized"}) + return + if parsed.path.startswith("/relays/"): + session_id = parsed.path.split("/relays/", 1)[1].strip("/") + with _lock: + info = _relays.get(session_id) + running = bool(info and _alive(int(info["pid"]))) + if info and not running: + _relays.pop(session_id, None) + if not running: + self._json(404, {"running": False, "session_id": session_id}) + return + self._json(200, {"running": True, "session_id": session_id, "pid": info["pid"], "path": info["path"]}) + return + self._json(404, {"error": "not found"}) + + def do_POST(self) -> None: + if not _authorized(self): + self._json(401, {"error": "unauthorized"}) + return + if urlparse(self.path).path != "/relays": + self._json(404, {"error": "not found"}) + return + length = int(self.headers.get("Content-Length") or 0) + try: + data = json.loads(self.rfile.read(length) or b"{}") + except json.JSONDecodeError: + self._json(400, {"error": "invalid json"}) + return + session_id = str(data.get("session_id") or "").strip() + path = str(data.get("path") or "").strip() + rtmps = str(data.get("rtmps") or "").strip() + if not session_id or not path or not rtmps.startswith("rtmps://"): + self._json(400, {"error": "session_id, path, rtmps required"}) + return + with _lock: + pid = _start_session(session_id, path, rtmps) + self._json(201, {"pid": pid, "session_id": session_id, "path": path}) + + def do_DELETE(self) -> None: + if not _authorized(self): + self._json(401, {"error": "unauthorized"}) + return + parsed = urlparse(self.path) + if not parsed.path.startswith("/relays/"): + self._json(404, {"error": "not found"}) + return + session_id = parsed.path.split("/relays/", 1)[1].strip("/") + with _lock: + _stop_session(session_id) + self._json(200, {"stopped": True, "session_id": session_id}) + + + def main() -> None: + host, port_s = LISTEN.rsplit(":", 1) + httpd = ThreadingHTTPServer((host, int(port_s)), Handler) + print(f"[relay-agent] listen {LISTEN} hls={HLS_BASE}") + httpd.serve_forever() + + + if __name__ == "__main__": + main() - path: /opt/stream-node/bootstrap.sh permissions: "0755" content: | diff --git a/native/android/app/src/androidTest/kotlin/com/matchlivetv/match_live_tv/E2EWizardFlowTest.kt b/native/android/app/src/androidTest/kotlin/com/matchlivetv/match_live_tv/E2EWizardFlowTest.kt index 44b9caa..0f484ab 100644 --- a/native/android/app/src/androidTest/kotlin/com/matchlivetv/match_live_tv/E2EWizardFlowTest.kt +++ b/native/android/app/src/androidTest/kotlin/com/matchlivetv/match_live_tv/E2EWizardFlowTest.kt @@ -65,6 +65,7 @@ class E2EWizardFlowTest { waitForText(s(R.string.wizard_step_title_transmission), timeoutMs = 45_000) waitForText(s(R.string.wizard_transmission_platform_title), timeoutMs = 30_000) selectYoutubeIfRequested() + selectUnlistedPrivacy() val onNetworkStep = { device.hasObject(By.text(s(R.string.wizard_network_test_start_label))) || device.hasObject(By.text("AVVIA TEST RETE")) || @@ -189,6 +190,12 @@ class E2EWizardFlowTest { if (!wantsYoutube()) return waitForAnyText("YouTube Live", timeoutMs = 20_000) tapClickableText("YouTube Live") + device.waitForIdle() + SystemClock.sleep(400) + } + + // Test su backend reali: mai pubblico (sito e YouTube). L'app espone solo pubblico / non in elenco. + private fun selectUnlistedPrivacy() { waitForAnyText( s(R.string.wizard_transmission_unlisted_label), "NON IN ELENCO", diff --git a/scripts/deploy/prod_relay_worker.sh b/scripts/deploy/prod_relay_worker.sh new file mode 100755 index 0000000..2159676 --- /dev/null +++ b/scripts/deploy/prod_relay_worker.sh @@ -0,0 +1,54 @@ +#!/usr/bin/env bash +# Fallback manuale: worker dedicato youtube_relay_ verso l'agent sul CPX. +# In produzione il Sidekiq home ascolta youtube_relay_cloud e chiama l'agent da solo; +# questo script serve solo se quel dispatch è spento o per debug. +# STREAM_NODE_SLUG=ingest-01 STREAM_NODE_RELAY_AGENT_URL=http://1.2.3.4:9100 \ +# bash scripts/deploy/prod_relay_worker.sh up +set -euo pipefail +ROOT="$(cd "$(dirname "$0")/../.." && pwd)" +INFRA="${MATCHLIVETV_INFRA:-$ROOT/infra}" +if [ -d /opt/matchlivetv/infra ]; then + INFRA=/opt/matchlivetv/infra +fi +SLUG="${STREAM_NODE_SLUG:?Imposta STREAM_NODE_SLUG (es. ingest-01)}" +ACTION="${1:-up}" +NAME="infra-sidekiq-relay-${SLUG}" +AGENT_URL="${STREAM_NODE_RELAY_AGENT_URL:-}" +AGENT_SECRET="${STREAM_NODE_AGENT_SECRET:-mediamtx_webhook_dev_secret}" +cd "$INFRA" +COMPOSE=(docker compose -f docker-compose.prod.yml --env-file .env) + +case "$ACTION" in + up) + docker rm -f "$NAME" >/dev/null 2>&1 || true + extra=() + if [[ -n "$AGENT_URL" ]]; then + extra+=( + -e STREAM_NODE_RELAY_AGENT_URL="$AGENT_URL" + -e STREAM_NODE_AGENT_SECRET="$AGENT_SECRET" + -e STREAM_NODE_LOCAL_RTMP_URL=rtmp://127.0.0.1:1935 + -e STREAM_NODE_LOCAL_HLS_URL=http://127.0.0.1:8888 + ) + fi + "${COMPOSE[@]}" run -d --name "$NAME" --no-deps \ + -e STREAM_NODE_SLUG="$SLUG" \ + -e YOUTUBE_RELAY_WORKER=1 \ + -e HOSTNAME="$SLUG" \ + -e RELAY_MAX_CONCURRENT="${RELAY_MAX_CONCURRENT:-2}" \ + "${extra[@]}" \ + sidekiq \ + bash -lc "bundle exec sidekiq -c \${RELAY_MAX_CONCURRENT:-2} -q youtube_relay_${SLUG} -e production" + echo "relay worker up name=${NAME} queue=youtube_relay_${SLUG} agent=${AGENT_URL:-local}" + ;; + down) + docker rm -f "$NAME" >/dev/null 2>&1 || true + echo "relay worker down name=${NAME}" + ;; + logs) + docker logs -f --tail=80 "$NAME" + ;; + *) + echo "Uso: STREAM_NODE_SLUG=ingest-01 STREAM_NODE_RELAY_AGENT_URL=http://IP:9100 $0 up|down|logs" >&2 + exit 1 + ;; +esac