glass beadian
aa5e262f3a
Self-hosted multi-platform broadcast newsroom for activist collectives: publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style column wall with hostile-content-safe rendering, live control room with Auto Director, bridged chat governance with polls and quorum, distributed sandboxed transcode workers, encrypted content-addressed archive, monthly analytics. MIT licensed. See README.md and docs/.
737 lines
37 KiB
Python
Executable file
737 lines
37 KiB
Python
Executable file
#!/usr/bin/python3
|
|
"""Whole-worker admission and operations for the official PeerTube runner."""
|
|
import argparse
|
|
import concurrent.futures
|
|
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import tomllib
|
|
import urllib.request
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
CONFIG = Path(os.environ.get('PT_WORKER_CONFIG', '/etc/pt-worker/worker.json'))
|
|
STATE = Path(os.environ.get('PT_STATE_DIR', '/var/lib/pt-worker/state'))
|
|
HOME = Path(os.environ.get('PT_WORKER_HOME', '/var/lib/pt-worker'))
|
|
RUNNER_CONFIG = HOME / '.config/peertube-runner-nodejs/default/config.toml'
|
|
VOD = ['vod-web-video-transcoding', 'vod-hls-transcoding', 'vod-audio-merge-transcoding']
|
|
spec = importlib.util.spec_from_file_location('qsv', ROOT / 'ffmpeg-qsv.py')
|
|
qsv = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(qsv)
|
|
|
|
|
|
def run(args, timeout=10):
|
|
return subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)
|
|
|
|
|
|
def atomic(path, data):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_name(path.name + '.new')
|
|
fd = os.open(temporary, os.O_CREAT | os.O_TRUNC | os.O_WRONLY, 0o600)
|
|
with os.fdopen(fd, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
f.write('\n')
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def read(path, default=None):
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except (OSError, ValueError):
|
|
return {} if default is None else default
|
|
|
|
|
|
def config():
|
|
value = read(CONFIG)
|
|
if not value.get('peertube_url') or not value.get('tailscale_peer'):
|
|
raise ValueError('worker.json needs peertube_url and tailscale_peer')
|
|
return value
|
|
|
|
|
|
def env():
|
|
return {**os.environ, 'HOME': str(HOME), 'PT_STATE_DIR': str(STATE),
|
|
'XDG_CONFIG_HOME': str(HOME / '.config'), 'XDG_CACHE_HOME': str(HOME / '.cache'),
|
|
'XDG_DATA_HOME': str(HOME / '.local/share'),
|
|
'FFMPEG_PATH': str(ROOT / 'ffmpeg-qsv.py'), 'FFPROBE_PATH': '/usr/bin/ffprobe',
|
|
'NODE_OPTIONS': '--require=' + str(ROOT / 'observe.cjs')}
|
|
|
|
|
|
def registered(cfg):
|
|
try:
|
|
data = tomllib.loads(RUNNER_CONFIG.read_text())
|
|
return any(r['url'].rstrip('/') == cfg['peertube_url'].rstrip('/') and
|
|
r['runnerName'] == cfg['name'] and r.get('runnerToken') for r in data.get('registeredInstances', []))
|
|
except (OSError, ValueError, KeyError):
|
|
return False
|
|
|
|
|
|
def ipc(action):
|
|
p = subprocess.run([config().get('node_binary', '/usr/bin/node'), str(ROOT / 'ipc.cjs'), action], env=env(),
|
|
capture_output=True, text=True, timeout=5)
|
|
if p.returncode:
|
|
raise RuntimeError('runner IPC unavailable')
|
|
return json.loads(p.stdout)
|
|
|
|
|
|
def network(cfg):
|
|
result = {'tailscale_ip': None, 'tailscale_path': 'unknown', 'latency_ms': None,
|
|
'network_ok': False, 'api_ok': False, 'peer_online': False}
|
|
try:
|
|
ts = json.loads(run(['tailscale', 'status', '--json'], 5).stdout)
|
|
result['tailscale_ip'] = next((a for a in ts.get('Self', {}).get('TailscaleIPs', []) if ':' not in a), None)
|
|
peers = list(ts.get('Peer', {}).values())
|
|
peer = next((p for p in peers if cfg['tailscale_peer'] in p.get('TailscaleIPs', []) or
|
|
p.get('HostName') == cfg['tailscale_peer'] or
|
|
p.get('DNSName', '').split('.')[0] == cfg['tailscale_peer']), None)
|
|
if ts.get('BackendState') != 'Running' or not peer or not peer.get('Online'):
|
|
return result
|
|
result['peer_online'] = True
|
|
p = run(['tailscale', 'ping', '--c=1', '--until-direct=false', '--timeout=3s', cfg['tailscale_peer']], 5)
|
|
text = p.stdout
|
|
if 'pong from' not in text:
|
|
return result
|
|
result['tailscale_path'] = 'DERP' if 'DERP(' in text else 'direct'
|
|
latency = re.search(r'in ([\d.]+)ms', text)
|
|
if latency:
|
|
result['latency_ms'] = float(latency[1])
|
|
started = time.monotonic()
|
|
with urllib.request.urlopen(cfg['peertube_url'].rstrip('/') + '/api/v1/config', timeout=4) as response:
|
|
version = json.load(response).get('serverVersion')
|
|
result['api_ok'] = version == cfg.get('peertube_version', '7.3.0')
|
|
result['api_latency_ms'] = (time.monotonic() - started) * 1000
|
|
result['network_ok'] = result['api_ok']
|
|
except (OSError, ValueError, subprocess.SubprocessError):
|
|
pass
|
|
return result
|
|
|
|
|
|
def transfer_probe(cfg, megabytes=8):
|
|
url = cfg.get('probe_url')
|
|
rates = {}
|
|
if not url:
|
|
return rates
|
|
payload = b'pt-worker-probe-' * ((megabytes * 1024**2 // 16) + 1)
|
|
payload = payload[:megabytes * 1024**2]
|
|
for direction in ('download', 'upload'):
|
|
begun, count = time.monotonic(), 0
|
|
try:
|
|
req = urllib.request.Request(url.rstrip('/') + ('/download?bytes=' + str(len(payload)) if direction == 'download' else '/upload'),
|
|
data=payload if direction == 'upload' else None,
|
|
method='PUT' if direction == 'upload' else 'GET')
|
|
with urllib.request.urlopen(req, timeout=30) as response:
|
|
if direction == 'upload':
|
|
count = json.load(response)['bytes_received']
|
|
else:
|
|
while chunk := response.read(1024**2):
|
|
count += len(chunk)
|
|
seconds = time.monotonic() - begun
|
|
qsv.event('transfer', direction=direction, bytes=count, seconds=seconds, ok=True, source='probe')
|
|
rates[direction + '_mbps'] = count * 8 / seconds / 1e6
|
|
except (OSError, ValueError, KeyError):
|
|
qsv.event('transfer', direction=direction, bytes=count, seconds=time.monotonic() - begun, ok=False, source='probe')
|
|
rates[direction + '_mbps'] = None
|
|
return rates
|
|
|
|
|
|
def sensors():
|
|
temperatures = []
|
|
for p in Path('/sys/class/thermal').glob('thermal_zone*/temp'):
|
|
try:
|
|
temperatures.append(float(p.read_text().strip()) / 1000)
|
|
except (OSError, ValueError):
|
|
pass
|
|
for p in Path('/sys/class/hwmon').glob('hwmon*/temp*_input'):
|
|
try:
|
|
temperatures.append(float(p.read_text().strip()) / 1000)
|
|
except (OSError, ValueError):
|
|
pass
|
|
ac, batteries = [], []
|
|
for p in Path('/sys/class/power_supply').glob('*'):
|
|
try:
|
|
kind = (p / 'type').read_text().strip()
|
|
if kind in ('Mains', 'USB', 'USB_C') and (p / 'online').exists():
|
|
ac.append((p / 'online').read_text().strip() == '1')
|
|
if kind == 'Battery':
|
|
batteries.append({'capacity': int((p / 'capacity').read_text()),
|
|
'status': (p / 'status').read_text().strip()})
|
|
except (OSError, ValueError):
|
|
pass
|
|
# Native servers with no battery have an implicit permanent power supply.
|
|
wsl = 'microsoft' in Path('/proc/sys/kernel/osrelease').read_text().lower()
|
|
on_ac = any(ac) if ac else (None if wsl or batteries else True)
|
|
throttles = {}
|
|
for p in Path('/sys/devices/system/cpu').glob('cpu*/thermal_throttle/*_throttle_count'):
|
|
try:
|
|
throttles[str(p)] = int(p.read_text())
|
|
except (OSError, ValueError):
|
|
pass
|
|
mem = {}
|
|
for line in Path('/proc/meminfo').read_text().splitlines():
|
|
key, value = line.split(':', 1)
|
|
if key in ('MemTotal', 'MemAvailable'):
|
|
mem[key] = int(value.strip().split()[0]) * 1024
|
|
return {'temperature_c': max(temperatures, default=None), 'temperatures_c': temperatures,
|
|
'on_ac': on_ac, 'battery': batteries, 'throttle_counts': throttles,
|
|
'ram_total_bytes': mem.get('MemTotal'), 'ram_available_bytes': mem.get('MemAvailable'),
|
|
'load_average': list(os.getloadavg()), 'wsl2': wsl}
|
|
|
|
|
|
def hardware():
|
|
cpu = next((l.split(':', 1)[1].strip() for l in Path('/proc/cpuinfo').read_text().splitlines()
|
|
if l.startswith('model name')), 'unknown')
|
|
gpu = []
|
|
for p in Path('/sys/class/drm').glob('renderD*/device'):
|
|
try:
|
|
gpu.append({'render': p.parent.name, 'vendor': (p / 'vendor').read_text().strip(),
|
|
'device': (p / 'device').read_text().strip(),
|
|
'driver': (p / 'driver').resolve().name})
|
|
except OSError:
|
|
pass
|
|
return {'cpu_model': cpu, 'gpu_model': gpu}
|
|
|
|
|
|
class Meter:
|
|
def __init__(self):
|
|
self.data = read(STATE / 'meter.json', {'offset': 0, 'jobs_completed': 0, 'jobs_failed': 0,
|
|
'transfer_failures': 0, 'active': {}})
|
|
|
|
def invalidate(self):
|
|
for key in ('download', 'upload'):
|
|
self.data.pop(key, None)
|
|
|
|
def update(self):
|
|
p = STATE / 'events.jsonl'
|
|
if not p.exists():
|
|
return
|
|
with p.open() as f:
|
|
if p.stat().st_size < self.data.get('offset', 0):
|
|
self.data['offset'] = 0
|
|
f.seek(self.data.get('offset', 0))
|
|
while True:
|
|
position = f.tell()
|
|
line = f.readline()
|
|
if not line:
|
|
break
|
|
if not line.endswith('\n'):
|
|
f.seek(position)
|
|
break
|
|
try:
|
|
self.consume(json.loads(line))
|
|
except (ValueError, KeyError, TypeError):
|
|
pass
|
|
self.data['offset'] = f.tell()
|
|
atomic(STATE / 'meter.json', self.data)
|
|
|
|
def consume(self, e):
|
|
kind, stamp = e['kind'], e['time']
|
|
if kind == 'transfer':
|
|
if not e['ok']:
|
|
self.data['transfer_failures'] = self.data.get('transfer_failures', 0) + 1
|
|
self.data['last_transfer_failure'] = stamp
|
|
elif e['bytes'] >= 65536 and e['seconds'] > 0:
|
|
direction = e['direction']
|
|
rate = e['bytes'] * 8 / e['seconds'] / 1e6
|
|
old = self.data.get(direction, {}).get('mbps', rate)
|
|
self.data[direction] = {'mbps': .3 * rate + .7 * old, 'time': stamp,
|
|
'bytes': e['bytes'], 'seconds': e['seconds']}
|
|
elif kind == 'gpu_failure':
|
|
self.data['gpu_failure_time'] = stamp
|
|
elif kind == 'encode':
|
|
self.data['last_encode'] = e
|
|
if e.get('job_uuid'):
|
|
self.data.setdefault('encodes', {})[e['job_uuid']] = e
|
|
elif kind == 'job_accept' and e['ok']:
|
|
self.data.setdefault('active', {})[e['uuid']] = stamp
|
|
elif kind in ('job_success', 'job_error', 'job_abort') and e['ok']:
|
|
begun = self.data.setdefault('active', {}).pop(e['uuid'], None)
|
|
if kind == 'job_success':
|
|
self.data['jobs_completed'] = self.data.get('jobs_completed', 0) + 1
|
|
elapsed = stamp - begun if begun else None
|
|
self.data['last_job_duration_seconds'] = elapsed
|
|
self.data['last_job_uuid'] = e['uuid']
|
|
encoded = self.data.setdefault('encodes', {}).pop(e['uuid'], {})
|
|
duration = encoded.get('duration')
|
|
self.data['last_job_realtime_factor'] = duration / elapsed if duration and elapsed else None
|
|
self.data['last_job_encode'] = encoded or None
|
|
elif kind == 'job_error':
|
|
self.data['jobs_failed'] = self.data.get('jobs_failed', 0) + 1
|
|
for path in (STATE / 'inputs').glob('*.json'):
|
|
if read(path).get('uuid') == e['uuid']:
|
|
path.unlink(missing_ok=True)
|
|
|
|
def rates(self, now, ttl):
|
|
return {f'recent_{k}_mbps': self.data.get(k, {}).get('mbps')
|
|
if now - self.data.get(k, {}).get('time', 0) < ttl else None for k in ('download', 'upload')}
|
|
|
|
|
|
class Admission:
|
|
def __init__(self, cfg):
|
|
self.cfg, self.good, self.bad = cfg, 0, 0
|
|
self.admitted, self.path, self.throttles = False, None, {}
|
|
|
|
def decide(self, sample, metric, control, now):
|
|
cfg = self.cfg
|
|
def refuse(state):
|
|
self.admitted, self.good = False, 0
|
|
return state
|
|
if self.path and sample['tailscale_path'] != self.path:
|
|
self.admitted = False
|
|
self.good = 0
|
|
metric.invalidate()
|
|
self.path = sample['tailscale_path']
|
|
rates = metric.rates(now, cfg.get('bandwidth_ttl_seconds', 900))
|
|
sample.update(rates)
|
|
if control.get('paused', True):
|
|
return refuse('PAUSED')
|
|
if not sample['registered']:
|
|
return refuse('PAUSED')
|
|
gpu_failed = metric.data.get('gpu_failure_time', 0) >= sample.get('gpu_tested_at', 0)
|
|
if not sample['gpu_accessible'] or not sample.get('gpu_ready') or gpu_failed:
|
|
return refuse('GPU_UNAVAILABLE')
|
|
if cfg.get('require_ac', True) and sample['on_ac'] is not True:
|
|
return refuse('DRAINING')
|
|
if sample['disk_free_bytes'] < cfg.get('min_disk_free_bytes', 20 * 1024**3):
|
|
return refuse('LOW_DISK')
|
|
rising = any(v > self.throttles.get(k, v) for k, v in sample['throttle_counts'].items())
|
|
self.throttles = sample['throttle_counts']
|
|
if rising or (sample['temperature_c'] is not None and sample['temperature_c'] >= cfg.get('max_temperature_c', 85)):
|
|
self.thermal_until = now + cfg.get('thermal_cooldown_seconds', 120)
|
|
if now < getattr(self, 'thermal_until', 0):
|
|
return refuse('THERMAL_THROTTLED')
|
|
if not sample['network_ok']:
|
|
self.admitted, self.good = False, 0
|
|
return 'OFFLINE'
|
|
calibrating = control.get('calibrate_until', 0) > now
|
|
factor = cfg.get('recovery_factor', 1.25) if not self.admitted else 1
|
|
bad = sample['latency_ms'] is None or sample['latency_ms'] > cfg.get('max_latency_ms', 150)
|
|
bad |= sample['tailscale_path'] == 'DERP' and not cfg.get('allow_derp', False)
|
|
if not calibrating:
|
|
bad |= any(rates[f'recent_{key}_mbps'] is None or rates[f'recent_{key}_mbps'] <
|
|
cfg.get(f'min_{key}_mbps', 10) * factor for key in ('download', 'upload'))
|
|
bad |= now - metric.data.get('last_transfer_failure', 0) < cfg.get('transfer_failure_cooldown_seconds', 60)
|
|
if bad:
|
|
self.good, self.bad = 0, self.bad + 1
|
|
if self.bad >= cfg.get('bad_samples', 2) or not self.admitted:
|
|
self.admitted = False
|
|
return 'NETWORK_DEGRADED'
|
|
else:
|
|
self.bad, self.good = 0, self.good + 1
|
|
if self.good >= cfg.get('good_samples', 3):
|
|
self.admitted = True
|
|
if not self.admitted:
|
|
return 'NETWORK_DEGRADED'
|
|
return 'BUSY' if sample['active_jobs'] else 'AVAILABLE'
|
|
|
|
|
|
class Utilization:
|
|
def __init__(self):
|
|
self.cpu, self.gpu, self.at = None, {}, time.monotonic()
|
|
|
|
def sample(self):
|
|
cpu = list(map(int, Path('/proc/stat').read_text().splitlines()[0].split()[1:9]))
|
|
gpu = {}
|
|
for p in Path('/proc').glob('[0-9]*'):
|
|
try:
|
|
if p.stat().st_uid == os.getuid() and (p / 'comm').read_text().strip() == 'ffmpeg':
|
|
for engine, value in qsv.engines(p.name).items():
|
|
gpu[(p.name, engine)] = value
|
|
except OSError:
|
|
pass
|
|
now = time.monotonic()
|
|
elapsed, result = now - self.at, {'cpu': None, 'gpu': None}
|
|
if self.cpu:
|
|
total = sum(cpu) - sum(self.cpu)
|
|
idle = cpu[3] + cpu[4] - self.cpu[3] - self.cpu[4]
|
|
result['cpu'] = (1 - idle / total) * 100 if total else None
|
|
if gpu:
|
|
by_engine = {}
|
|
for key, value in gpu.items():
|
|
if key in self.gpu:
|
|
delta = max(0, value - self.gpu[key])
|
|
by_engine[key[1]] = by_engine.get(key[1], 0) + delta / (elapsed * 1e9) * 100
|
|
result['gpu'] = by_engine or None
|
|
self.cpu, self.gpu, self.at = cpu, gpu, now
|
|
return result
|
|
|
|
|
|
def log_runner(pipe):
|
|
# Upstream logs contain job tokens and sometimes runner tokens: retain only safe milestones.
|
|
patterns = [r'Processing job of type ([\w-]+): ([a-f\d-]{36})',
|
|
r'Finished processing job of type ([\w-]+): ([a-f\d-]{36})']
|
|
for line in pipe:
|
|
for pattern in patterns:
|
|
match = re.search(pattern, line)
|
|
if match:
|
|
qsv.event('runner_milestone', phase='completed' if 'Finished' in pattern else 'processing',
|
|
job_type=match[1], uuid=match[2])
|
|
if 'Cannot process job' in line:
|
|
qsv.event('runner_error', reason='process_job_failed')
|
|
|
|
|
|
def daemon():
|
|
cfg = config()
|
|
STATE.mkdir(parents=True, exist_ok=True)
|
|
lock = (STATE / 'daemon.lock').open('w')
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
meter, admission, utilization = Meter(), Admission(cfg), Utilization()
|
|
child, draining, stopping, restart_after, last_probe, previous_path = None, False, False, 0, 0, None
|
|
def stop(*_):
|
|
nonlocal stopping
|
|
stopping = True
|
|
signal.signal(signal.SIGTERM, stop)
|
|
signal.signal(signal.SIGINT, stop)
|
|
hw = hardware()
|
|
proof = read(STATE / 'qsv.json')
|
|
boot_id = Path('/proc/sys/kernel/random/boot_id').read_text().strip()
|
|
if proof.get('boot_id') != boot_id and os.access(cfg.get('render_device', '/dev/dri/renderD128'), os.R_OK | os.W_OK):
|
|
self_test()
|
|
while True:
|
|
now = time.time()
|
|
meter.update()
|
|
control = read(STATE / 'control.json', {'paused': True})
|
|
if stopping:
|
|
control['paused'] = True
|
|
if child and child.poll() is not None:
|
|
restart_after = now + 15
|
|
child, draining = None, False
|
|
sample = {**network(cfg), **sensors(), **hw, 'hostname': cfg['name'], 'updated_at': now,
|
|
'registered': registered(cfg), 'registration_state': 'registered' if registered(cfg) else 'unregistered',
|
|
'concurrency': cfg.get('concurrency', 2), 'active_jobs': 0 if not child else None,
|
|
'qsv': read(STATE / 'qsv.json'), 'draining': draining}
|
|
gpu_health = read(Path(cfg.get('gpu_health_file', str(STATE / 'qsv.json'))))
|
|
sample.update(gpu_ready=gpu_health.get('hardware_video_ok', False), gpu_tested_at=gpu_health.get('tested_at', 0))
|
|
device = cfg.get('gpu_device', cfg.get('render_device', '/dev/dri/renderD128'))
|
|
sample['gpu_accessible'] = os.access(device, os.R_OK | os.W_OK)
|
|
disk = os.statvfs(HOME)
|
|
sample['disk_free_bytes'] = disk.f_bavail * disk.f_frsize
|
|
ipc_ok = not child
|
|
if child:
|
|
try:
|
|
sample.update(ipc('list-jobs'))
|
|
ipc_ok = True
|
|
except (OSError, ValueError, subprocess.SubprocessError, RuntimeError):
|
|
pass
|
|
if previous_path and previous_path != sample['tailscale_path']:
|
|
meter.invalidate()
|
|
last_probe = 0
|
|
admission.path = None
|
|
admission.admitted, admission.good = False, 0
|
|
previous_path = sample['tailscale_path']
|
|
probe_due = cfg.get('probe_url') and not control.get('paused', True) and sample['network_ok'] and \
|
|
sample['on_ac'] is True and sample['tailscale_path'] == 'direct' and sample['active_jobs'] == 0 and \
|
|
now - last_probe >= cfg.get('probe_interval_seconds', 300)
|
|
if probe_due and child and not draining:
|
|
try:
|
|
result = subprocess.run([cfg['runner_binary'], 'graceful-shutdown'], env=env(),
|
|
capture_output=True, text=True, timeout=5)
|
|
draining = result.returncode == 0
|
|
except (OSError, subprocess.SubprocessError):
|
|
pass
|
|
if probe_due and not child:
|
|
transfer_probe(cfg)
|
|
last_probe = now
|
|
meter.update()
|
|
sample.update(network(cfg))
|
|
sample.update(sensors())
|
|
if sample['tailscale_path'] != previous_path:
|
|
meter.invalidate()
|
|
admission.admitted, admission.good, admission.path = False, 0, None
|
|
last_probe = 0
|
|
now = time.time()
|
|
desired = admission.decide(sample, meter, control, now)
|
|
if not ipc_ok:
|
|
desired = 'OFFLINE'
|
|
allow = desired in ('AVAILABLE', 'BUSY') and not stopping
|
|
if child and not allow and not draining:
|
|
try:
|
|
# SIGTERM upstream aborts jobs. Use its official graceful-shutdown command instead.
|
|
result = subprocess.run([cfg['runner_binary'], 'graceful-shutdown'], env=env(),
|
|
capture_output=True, text=True, timeout=5)
|
|
draining = result.returncode == 0
|
|
if draining:
|
|
qsv.event('drain', reason=desired)
|
|
except (OSError, subprocess.SubprocessError):
|
|
pass
|
|
if not child and allow and now >= restart_after:
|
|
child = subprocess.Popen([cfg['runner_binary'], 'server', *sum([['--enable-job', j] for j in VOD], [])],
|
|
env=env(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
|
|
start_new_session=True)
|
|
threading.Thread(target=log_runner, args=(child.stdout,), daemon=True).start()
|
|
sample.update({'status': 'DRAINING' if draining and desired in ('AVAILABLE', 'BUSY') else desired,
|
|
'draining': draining, 'runner_running': child is not None,
|
|
'qsv_h264_ok': sample['qsv'].get('qsv_h264_ok'),
|
|
'qsv_hevc_ok': sample['qsv'].get('qsv_hevc_ok'),
|
|
'util': utilization.sample(),
|
|
**{k: meter.data.get(k) for k in ('jobs_completed', 'jobs_failed', 'transfer_failures',
|
|
'last_job_uuid', 'last_job_duration_seconds', 'last_job_realtime_factor', 'last_job_encode', 'last_encode')}})
|
|
# Null is intentional until the corresponding measurement has been made.
|
|
atomic(STATE / 'status.json', sample)
|
|
if stopping and child is None:
|
|
return
|
|
# After suspend, wall-clock gaps force a fresh path/throughput evaluation.
|
|
before = time.time()
|
|
time.sleep(cfg.get('poll_seconds', 5))
|
|
if time.time() - before > cfg.get('poll_seconds', 5) * 3:
|
|
meter.invalidate()
|
|
admission.admitted, admission.good = False, 0
|
|
|
|
|
|
def measured(command, log):
|
|
started = time.monotonic()
|
|
maximum, samples = {}, []
|
|
with log.open('w') as f:
|
|
p = subprocess.Popen(command, stdout=f, stderr=subprocess.STDOUT)
|
|
while p.poll() is None:
|
|
for k, v in qsv.engines(p.pid).items():
|
|
maximum[k] = max(maximum.get(k, 0), v)
|
|
samples.append(sensors())
|
|
if time.monotonic() - started > 180:
|
|
p.kill()
|
|
p.wait()
|
|
break
|
|
time.sleep(.1)
|
|
return {'returncode': p.returncode, 'seconds': time.monotonic() - started,
|
|
'engine_ns': maximum, 'max_temperature_c': max((s['temperature_c'] for s in samples
|
|
if s['temperature_c'] is not None), default=None),
|
|
'min_available_ram_bytes': min((s['ram_available_bytes'] for s in samples), default=None)}
|
|
|
|
|
|
def self_test():
|
|
cfg = config()
|
|
device = cfg.get('render_device', '/dev/dri/renderD128')
|
|
STATE.mkdir(parents=True, exist_ok=True)
|
|
report = {**hardware(), 'tested_at': time.time(), 'boot_id': Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
|
|
'qsv_h264_ok': False, 'qsv_hevc_ok': False,
|
|
'av1_decode_ok': None, 'av1_encode_ok': None, 'tests': {}, 'device': device}
|
|
diagnostics = STATE / 'self-test'
|
|
diagnostics.mkdir(exist_ok=True)
|
|
for name, command in [('ffmpeg-version', ['/usr/bin/ffmpeg', '-version']),
|
|
('hwaccels', ['/usr/bin/ffmpeg', '-hwaccels']),
|
|
('encoders', ['/usr/bin/ffmpeg', '-encoders']),
|
|
('vainfo', ['vainfo', '--display', 'drm', '--device', device])]:
|
|
try:
|
|
p = run(command, 20)
|
|
(diagnostics / (name + '.log')).write_text(p.stdout + p.stderr)
|
|
report[name + '_exit'] = p.returncode
|
|
except (OSError, subprocess.SubprocessError):
|
|
report[name + '_exit'] = None
|
|
if not os.access(device, os.R_OK | os.W_OK):
|
|
report['reason'] = 'render device missing or inaccessible; hardware tests did not run'
|
|
atomic(STATE / 'qsv.json', report)
|
|
print(json.dumps(report, indent=2))
|
|
return 1
|
|
with tempfile.TemporaryDirectory(prefix='selftest-', dir=STATE) as directory:
|
|
tmp = Path(directory)
|
|
source = tmp / 'source.mp4'
|
|
# CPU x264 is used ONLY to create synthetic test input, outside the runner.
|
|
p = run(['/usr/bin/ffmpeg', '-v', 'error', '-f', 'lavfi', '-i', 'testsrc2=size=1920x1080:rate=30',
|
|
'-f', 'lavfi', '-i', 'sine=frequency=1000:sample_rate=48000', '-t', '10', '-c:v', 'libx264',
|
|
'-threads', '2', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', '-c:a', 'aac', str(source)], 90)
|
|
if p.returncode:
|
|
report['reason'] = 'test fixture creation failed'
|
|
else:
|
|
base = ['/usr/bin/ffmpeg', '-v', 'verbose', '-init_hw_device', f'vaapi=va:{device}',
|
|
'-init_hw_device', 'qsv=qs@va', '-filter_hw_device', 'qs', '-hwaccel', 'qsv',
|
|
'-hwaccel_output_format', 'qsv', '-c:v', 'h264_qsv', '-i', str(source),
|
|
'-vf', 'scale_qsv=w=1280:h=720:format=nv12', '-an']
|
|
for codec, key in [('h264', 'qsv_h264_ok'), ('hevc', 'qsv_hevc_ok'), ('av1', 'av1_encode_ok')]:
|
|
out = tmp / (codec + '.mp4')
|
|
result = measured([*base, '-c:v', codec + '_qsv', '-b:v', '2M', str(out)], diagnostics / (codec + '.log'))
|
|
result['hardware_engine_evidence'] = any(v > 0 for k, v in result['engine_ns'].items() if k.startswith('video'))
|
|
report['tests'][codec] = result
|
|
report[key] = result['returncode'] == 0 and result['hardware_engine_evidence']
|
|
if result['returncode'] == 0:
|
|
report['tests'][codec]['output_probe'] = qsv.probe(str(out))
|
|
if report['qsv_hevc_ok']:
|
|
command = base[:base.index('-c:v')] + ['-re', '-c:v', 'hevc_qsv', '-i', str(tmp / 'hevc.mp4'),
|
|
'-vf', 'scale_qsv=w=640:h=360:format=nv12', '-an', '-c:v', 'h264_qsv', '-f', 'null', '-']
|
|
result = measured(command, diagnostics / 'hevc-decode.log')
|
|
report['tests']['hevc_decode'] = result
|
|
report['qsv_hevc_decode_ok'] = result['returncode'] == 0 and any(v > 0 for k, v in result['engine_ns'].items() if k.startswith('video'))
|
|
av1_source = tmp / 'av1-source.mkv'
|
|
p = run(['/usr/bin/ffmpeg', '-v', 'error', '-f', 'lavfi', '-i', 'testsrc2=size=640x360:rate=30',
|
|
'-t', '2', '-c:v', 'libaom-av1', '-threads', '2', '-cpu-used', '8', str(av1_source)], 90)
|
|
if p.returncode == 0:
|
|
command = base[:base.index('-c:v')] + ['-re', '-c:v', 'av1_qsv', '-i', str(av1_source),
|
|
'-vf', 'scale_qsv=w=640:h=360:format=nv12', '-an', '-c:v', 'h264_qsv', '-f', 'null', '-']
|
|
result = measured(command, diagnostics / 'av1-decode.log')
|
|
report['tests']['av1_decode'] = result
|
|
report['av1_decode_ok'] = result['returncode'] == 0 and any(v > 0 for k, v in result['engine_ns'].items() if k.startswith('video'))
|
|
# Exercise the adapter itself; compare with the direct QSV tests.
|
|
command = [str(ROOT / 'ffmpeg-qsv.py'), '-i', str(source), '-vcodec', 'libx264',
|
|
'-vf', 'scale=w=-2:h=720', '-pix_fmt', 'yuv420p', '-preset', 'veryfast',
|
|
'-maxrate:v', '2000000', '-bufsize:v', '4000000', '-b_strategy', '1', '-bf', '16',
|
|
'-g:v', '60', '-r', '30', '-c:a', 'aac', str(tmp / 'adapter.mp4')]
|
|
p = subprocess.run(command, env=env(), capture_output=True, text=True, timeout=90)
|
|
(diagnostics / 'adapter.log').write_text(p.stderr)
|
|
report['adapter_exit'] = p.returncode
|
|
report['qsv_h264_ok'] &= p.returncode == 0
|
|
# The admission controller compares this timestamp to failures emitted by earlier tests.
|
|
report['tested_at'] = time.time()
|
|
report['hardware_video_ok'] = report['qsv_h264_ok']
|
|
report['backend'] = 'intel_qsv'
|
|
atomic(STATE / 'qsv.json', report)
|
|
print(json.dumps(report, indent=2))
|
|
return 0 if report['qsv_h264_ok'] else 1
|
|
|
|
|
|
def network_test(args):
|
|
"""Real HTTP throughput against explicit disposable operator-supplied endpoints."""
|
|
cfg = config()
|
|
report = network(cfg)
|
|
if not args.download_url or not args.upload_url:
|
|
if cfg.get('probe_url'):
|
|
rates = transfer_probe(cfg, min(args.megabytes, 16))
|
|
print(json.dumps({**report, **rates, 'throughput_measured': all(v is not None for v in rates.values())}, indent=2))
|
|
return 0 if all(v is not None for v in rates.values()) else 1
|
|
print(json.dumps({**report, 'throughput_measured': False,
|
|
'reason': 'supply --download-url and --upload-url for disposable fixtures; API ping is not bandwidth'}, indent=2))
|
|
return 1
|
|
rates = {}
|
|
for direction, url in [('download', args.download_url), ('upload', args.upload_url)]:
|
|
# Test URLs must remain within the same PeerTube/storage route.
|
|
if urllib.parse.urlparse(url).hostname != urllib.parse.urlparse(cfg['peertube_url']).hostname:
|
|
raise ValueError('test URL must use configured PeerTube host')
|
|
count, begun = 0, time.monotonic()
|
|
try:
|
|
req = urllib.request.Request(url)
|
|
if direction == 'upload':
|
|
req = urllib.request.Request(url, data=os.urandom(args.megabytes * 1024**2), method='PUT')
|
|
count = args.megabytes * 1024**2
|
|
with urllib.request.urlopen(req, timeout=60) as response:
|
|
while chunk := response.read(1024**2):
|
|
if direction == 'download':
|
|
count += len(chunk)
|
|
seconds = time.monotonic() - begun
|
|
qsv.event('transfer', direction=direction, bytes=count, seconds=seconds, ok=True)
|
|
rates[direction + '_mbps'] = count * 8 / seconds / 1e6
|
|
except OSError:
|
|
qsv.event('transfer', direction=direction, bytes=count, seconds=time.monotonic() - begun, ok=False)
|
|
rates[direction + '_mbps'] = None
|
|
print(json.dumps({**report, **rates, 'throughput_measured': all(v is not None for v in rates.values())}, indent=2))
|
|
return 0 if all(v is not None for v in rates.values()) else 1
|
|
|
|
|
|
def benchmark(args):
|
|
status = read(STATE / 'status.json')
|
|
if status.get('runner_running'):
|
|
raise ValueError('drain worker and wait for runner_running=false before benchmarking')
|
|
source = Path(args.source).resolve()
|
|
probe = qsv.probe(str(source))
|
|
duration = float(probe.get('format', {}).get('duration', 0))
|
|
records = []
|
|
with tempfile.TemporaryDirectory(prefix='benchmark-', dir=STATE) as directory:
|
|
tmp = Path(directory)
|
|
for concurrency in (1, 2, 3, 4):
|
|
def job(index):
|
|
out = tmp / f'{concurrency}-{index}.mp4'
|
|
cmd = [str(ROOT / 'ffmpeg-qsv.py'), '-i', str(source), '-vcodec', 'libx264',
|
|
'-vf', 'scale=w=-2:h=720', '-pix_fmt', 'yuv420p', '-maxrate:v', '2000000',
|
|
'-bufsize:v', '4000000', '-r', '30', '-an', str(out)]
|
|
start = time.monotonic()
|
|
p = subprocess.run(cmd, env=env(), capture_output=True, text=True, timeout=600)
|
|
elapsed = time.monotonic() - start
|
|
out.unlink(missing_ok=True)
|
|
return {'seconds': elapsed, 'realtime_factor': duration / elapsed,
|
|
'fps': duration * 30 / elapsed, 'returncode': p.returncode}
|
|
start, samples, util_samples = time.monotonic(), [], []
|
|
utilization = Utilization()
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
|
|
futures = [executor.submit(job, i) for i in range(concurrency)]
|
|
while not all(f.done() for f in futures):
|
|
samples.append(sensors())
|
|
util_samples.append(utilization.sample())
|
|
time.sleep(.5)
|
|
jobs = [f.result() for f in futures]
|
|
wall = time.monotonic() - start
|
|
records.append({'concurrency': concurrency, 'wall_seconds': wall, 'jobs': jobs,
|
|
'aggregate_media_seconds_per_second': concurrency * duration / wall,
|
|
'max_temperature_c': max((s['temperature_c'] for s in samples
|
|
if s['temperature_c'] is not None), default=None),
|
|
'min_available_ram_bytes': min((s['ram_available_bytes'] for s in samples), default=None),
|
|
'utilization_samples': util_samples,
|
|
'throttle_counts_start': samples[0]['throttle_counts'] if samples else {},
|
|
'throttle_counts_end': samples[-1]['throttle_counts'] if samples else {},
|
|
'recent_download_mbps': None, 'recent_upload_mbps': None})
|
|
report = {'scope': 'local encode diagnostic; choose production concurrency from real end-to-end jobs',
|
|
'source_duration': duration, 'results': records, 'production_concurrency_selected': None}
|
|
atomic(STATE / 'benchmark.json', report)
|
|
print(json.dumps(report, indent=2))
|
|
return 0 if all(j['returncode'] == 0 for r in records for j in r['jobs']) else 1
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
sub = parser.add_subparsers(dest='command', required=True)
|
|
for name in ('daemon', 'status', 'start', 'stop', 'restart', 'drain', 'self-test'):
|
|
p = sub.add_parser(name)
|
|
if name == 'start':
|
|
p.add_argument('--calibrate', action='store_true', help='explicit 10-minute initial job measurement window')
|
|
if name == 'status':
|
|
p.add_argument('--prometheus', action='store_true')
|
|
p = sub.add_parser('network-test')
|
|
p.add_argument('--download-url')
|
|
p.add_argument('--upload-url')
|
|
p.add_argument('--megabytes', type=int, default=16)
|
|
p = sub.add_parser('benchmark')
|
|
p.add_argument('--source', required=True)
|
|
args = parser.parse_args()
|
|
if args.command == 'daemon':
|
|
return daemon()
|
|
if args.command == 'status':
|
|
empty = {k: None for k in ('tailscale_ip', 'tailscale_path', 'registration_state', 'cpu_model', 'gpu_model',
|
|
'qsv_h264_ok', 'qsv_hevc_ok', 'active_jobs', 'concurrency', 'util', 'temperature_c', 'temperatures_c',
|
|
'disk_free_bytes', 'recent_download_mbps', 'recent_upload_mbps', 'jobs_completed', 'jobs_failed',
|
|
'last_job_duration_seconds', 'last_job_realtime_factor')}
|
|
value = read(STATE / 'status.json', {**empty, 'hostname': socket.gethostname(), 'status': 'OFFLINE'})
|
|
if time.time() - value.get('updated_at', 0) > 60:
|
|
value['status'] = 'OFFLINE'
|
|
value['stale'] = True
|
|
if args.prometheus:
|
|
print('pt_worker_available ' + str(int(value.get('status') in ('AVAILABLE', 'BUSY'))))
|
|
for key in ('active_jobs', 'concurrency', 'disk_free_bytes', 'temperature_c', 'recent_download_mbps',
|
|
'recent_upload_mbps', 'jobs_completed', 'jobs_failed'):
|
|
if isinstance(value.get(key), (int, float)):
|
|
print('pt_worker_' + key + ' ' + str(value[key]))
|
|
else:
|
|
print(json.dumps(value, indent=2))
|
|
return 0
|
|
if args.command in ('start', 'stop', 'restart', 'drain'):
|
|
control = {'paused': args.command in ('stop', 'drain'),
|
|
'calibrate_until': time.time() + 600 if getattr(args, 'calibrate', False) else 0}
|
|
atomic(STATE / 'control.json', control)
|
|
if args.command == 'restart':
|
|
# Drain first; the controller restarts the child after it exits, preserving registration.
|
|
try:
|
|
subprocess.run([config()['runner_binary'], 'graceful-shutdown'], env=env(),
|
|
capture_output=True, timeout=5, check=True)
|
|
except (OSError, subprocess.SubprocessError):
|
|
pass
|
|
print(json.dumps({'requested': args.command, **control}))
|
|
return 0
|
|
if args.command == 'self-test':
|
|
if read(STATE / 'status.json').get('runner_running'):
|
|
raise ValueError('drain and wait for runner_running=false before self-test')
|
|
return self_test()
|
|
if args.command == 'network-test':
|
|
return network_test(args)
|
|
if args.command == 'benchmark':
|
|
return benchmark(args)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
sys.exit(main() or 0)
|
|
except (ValueError, RuntimeError, OSError, subprocess.SubprocessError) as error:
|
|
print('pt-worker: ' + str(error), file=sys.stderr)
|
|
sys.exit(1)
|