#!/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)