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/.
459 lines
21 KiB
Python
459 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""pt-worker admission controller, native Windows port (validator for the design
|
|
in windows-worker-result.md).
|
|
|
|
Ported from infra/peertube-workers/pt-worker.py: identical Meter / Admission
|
|
state machine and policy defaults (AC-only, 20 GiB reserve, 85 C ceiling,
|
|
10 Mbps floors with 1.25x recovery factor, DERP disallowed, 3-good/2-bad
|
|
hysteresis, 120 s thermal cooldown, 60 s transfer-failure cooldown, 900 s
|
|
bandwidth TTL, 5 s poll). Windows deltas:
|
|
- power/battery via GetSystemPowerStatus (ctypes; None == unknown == drain);
|
|
Modern Standby is mitigated with `powercfg /change standby-timeout-ac 0`
|
|
during install, not here.
|
|
- temperatures via MSAcpi_ThermalZoneTemperature; unsupported sensors yield
|
|
None (unsupported measurements are never invented).
|
|
- GPU encode telemetry from PDH GPU Engine (engtype_VideoEncode) counters;
|
|
DRM fdinfo does not exist on Windows. Node >=18.20/20.12: `os.loadavg()`
|
|
returns [0,0,0] on Windows and is not emitted.
|
|
- cross-process daemon lock is a win32 mutex via msvcrt (fcntl is POSIX).
|
|
- drain is delivered through the official runner named-pipe IPC.
|
|
- hardware re-proof triggers on driver_version change in addition to boot_id
|
|
(Windows Update can silently change graphics drivers).
|
|
|
|
Runs on CPython >= 3.11 with psutil (optional: improves disk/memory collection).
|
|
Emitting a status JSON is safe to run standalone on Windows or elsewhere.
|
|
"""
|
|
import argparse
|
|
import ctypes
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import time
|
|
import tomllib
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(r'C:\pt-worker')
|
|
CONFIG = ROOT / 'state' / 'worker.json'
|
|
STATE = ROOT / 'state'
|
|
RUNNER_CONFIG = Path(os.environ.get('APPDATA', str(Path.home() / 'AppData' / 'Roaming'))) / 'peertube-runner-nodejs' / 'Config' / 'config.toml'
|
|
|
|
DEFAULTS = {
|
|
'require_ac': True, 'min_disk_free_bytes': 20 * 1024 ** 3, 'max_temperature_c': 85,
|
|
'min_download_mbps': 10, 'min_upload_mbps': 10, 'max_latency_ms': 150,
|
|
'allow_derp': False, 'recovery_factor': 1.25, 'good_samples': 3, 'bad_samples': 2,
|
|
'bandwidth_ttl_seconds': 900, 'transfer_failure_cooldown_seconds': 60,
|
|
'thermal_cooldown_seconds': 120, 'poll_seconds': 5,
|
|
}
|
|
# Last-resort resource gates that hold even while admission policy is disabled.
|
|
# Same thresholds as the Linux installer defaults (15 GiB RAM floor for a
|
|
# 2-concurrency QSV worker, 6 GiB min per-job scratch); no CPU fallback exists.
|
|
HARD_FLOORS = {'ram_total_bytes': 15 * 1024 ** 3, 'min_scratch_bytes': 6 * 1024 ** 3}
|
|
|
|
|
|
def run(args, timeout=10):
|
|
return subprocess.run(args, capture_output=True, text=True, timeout=timeout)
|
|
|
|
|
|
def read(path, default=None):
|
|
try:
|
|
return json.loads(Path(path).read_text(encoding='utf8'))
|
|
except (OSError, ValueError):
|
|
return default
|
|
|
|
|
|
def config():
|
|
value = read(CONFIG)
|
|
if not value or not value.get('peertube_url') or not value.get('tailscale_peer'):
|
|
raise ValueError('worker.json needs peertube_url and tailscale_peer')
|
|
return {**DEFAULTS, **value}
|
|
|
|
|
|
def ipc(action, node_binary='node'):
|
|
"""Official runner IPC via ipc-win.cjs; the runner must be running in server mode."""
|
|
env = {**os.environ, 'PT_STATE_DIR': str(STATE)}
|
|
p = run([node_binary, str(ROOT / 'bin' / 'ipc-win.cjs'), action], 15)
|
|
if p.returncode:
|
|
raise RuntimeError('runner IPC unavailable')
|
|
return json.loads(p.stdout)
|
|
|
|
|
|
def registered(cfg):
|
|
try:
|
|
data = tomllib.loads(RUNNER_CONFIG.read_text(encoding='utf8'))
|
|
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 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)
|
|
if 'pong from' not in p.stdout:
|
|
return result
|
|
result['tailscale_path'] = 'DERP' if 'DERP(' in p.stdout else 'direct'
|
|
latency = re.search(r'in ([\d.]+)ms', p.stdout)
|
|
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['network_ok'] = result['api_ok']
|
|
except (OSError, ValueError, subprocess.SubprocessError, urllib.error.URLError):
|
|
pass
|
|
return result
|
|
|
|
|
|
def transfer_probe(cfg, megabytes=8):
|
|
"""Bounded probe against pt-worker-netprobe (same server-side probe as Linux)."""
|
|
url = cfg.get('probe_url')
|
|
rates = {}
|
|
if not url:
|
|
return rates
|
|
payload = os.urandom(megabytes * 1024 * 1024)
|
|
for direction, request in (('download', urllib.request.Request(url.rstrip('/') + '/download',
|
|
data=None, method='GET')),
|
|
('upload', None)):
|
|
try:
|
|
started = time.monotonic()
|
|
if direction == 'download':
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
seen = 0
|
|
while len(response.read(1 << 20)):
|
|
seen += 1 << 20
|
|
if seen >= megabytes * 1024 * 1024:
|
|
break
|
|
seconds = time.monotonic() - started
|
|
if seen < 65536:
|
|
continue
|
|
else:
|
|
req = urllib.request.Request(url.rstrip('/') + '/upload', data=payload, method='POST')
|
|
urllib.request.urlopen(req, timeout=30).read()
|
|
seconds = time.monotonic() - started
|
|
rates[direction] = {'mbps': megabytes * 8 / seconds, 'time': time.time(),
|
|
'bytes': megabytes * 1024 * 1024, 'seconds': seconds}
|
|
except (OSError, urllib.error.URLError, ValueError):
|
|
rates[direction + '_failed'] = True
|
|
return rates
|
|
|
|
|
|
def get_system_power_status():
|
|
"""Win32 GetSystemPowerStatus; dict or None when unsupported/unknown."""
|
|
class PLS(ctypes.Structure):
|
|
_fields_ = [('ACLineStatus', ctypes.c_byte), ('BatteryFlag', ctypes.c_byte),
|
|
('BatteryLifePercent', ctypes.c_byte), ('Reserved1', ctypes.c_byte),
|
|
('BatteryLifeTime', ctypes.c_uint32), ('BatteryFullLifeTime', ctypes.c_uint32)]
|
|
status = PLS()
|
|
if not ctypes.windll.kernel32.GetSystemPowerStatus(ctypes.byref(status)):
|
|
return None
|
|
ac = {0: False, 1: True}.get(status.ACLineStatus) # 255 == unknown -> None
|
|
percentage = status.BatteryLifePercent
|
|
return {'on_ac': ac, 'battery_flag': status.BatteryFlag,
|
|
'battery_percent': None if percentage in (0, 255) else percentage}
|
|
|
|
|
|
def sensors():
|
|
result = {'temperature_c': None, 'temperatures_c': [], 'on_ac': None, 'battery': [],
|
|
'throttle_counts': {}, 'ram_total_bytes': None, 'ram_available_bytes': None,
|
|
'load_average': None}
|
|
try:
|
|
p = run(['powershell', '-NoProfile', '-NonInteractive', '-Command',
|
|
'(Get-CimInstance -ClassName MSAcpi_ThermalZoneTemperature -ErrorAction Stop | '
|
|
'Measure-Object -Property CurrentTemperature -Maximum).Maximum'], 12)
|
|
if p.returncode == 0:
|
|
value = float(p.stdout.strip()) / 10.0 - 273.15
|
|
if 0 < value < 120:
|
|
result['temperature_c'] = round(value, 1)
|
|
result['temperatures_c'] = [result['temperature_c']]
|
|
except (OSError, ValueError, subprocess.SubprocessError):
|
|
pass
|
|
power = get_system_power_status()
|
|
if power:
|
|
result['on_ac'] = power['on_ac']
|
|
if power['battery_percent'] is not None:
|
|
result['battery'] = [{'capacity': power['battery_percent'],
|
|
'status': {0: 'unknown', 1: 'discharging', 2: 'on_ac'}.get(power['battery_flag'], 'other')}]
|
|
try:
|
|
import ctypes.wintypes # noqa: F401 (MEMORYSTATUSEX needs no extra types)
|
|
class MEMORYSTATUSEX(ctypes.Structure):
|
|
_fields_ = [('dwLength', ctypes.c_uint32), ('dwMemoryLoad', ctypes.c_uint32),
|
|
('ullTotalPhys', ctypes.c_uint64), ('ullAvailPhys', ctypes.c_uint64),
|
|
('ullTotalPageFile', ctypes.c_uint64), ('ullAvailPageFile', ctypes.c_uint64),
|
|
('ullTotalVirtual', ctypes.c_uint64), ('ullAvailVirtual', ctypes.c_uint64),
|
|
('ullAvailExtendedVirtual', ctypes.c_uint64)]
|
|
memory = MEMORYSTATUSEX()
|
|
memory.dwLength = ctypes.sizeof(MEMORYSTATUSEX)
|
|
if ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(memory)):
|
|
result['ram_total_bytes'] = memory.ullTotalPhys
|
|
result['ram_available_bytes'] = memory.ullAvailPhys
|
|
except (OSError, AttributeError, ValueError):
|
|
pass
|
|
return result
|
|
|
|
|
|
def disk_free(path=str(ROOT)):
|
|
try:
|
|
free = ctypes.c_ulonglong()
|
|
if ctypes.windll.kernel32.GetDiskFreeSpaceExW(ctypes.c_wchar_p(path), None, None, ctypes.byref(free)):
|
|
return free.value
|
|
except (OSError, AttributeError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def gpu_identity():
|
|
try:
|
|
p = run(['powershell', '-NoProfile', '-NonInteractive', '-Command',
|
|
'Get-CimInstance Win32_VideoController | Select-Object Name,DriverVersion,Status | ConvertTo-Json'], 12)
|
|
data = json.loads(p.stdout or '[]')
|
|
if isinstance(data, dict):
|
|
data = [data]
|
|
return data
|
|
except (OSError, ValueError, subprocess.SubprocessError):
|
|
return []
|
|
|
|
|
|
class Meter:
|
|
"""Byte-identical semantics to the Linux controller's Meter."""
|
|
|
|
def __init__(self):
|
|
self.data = read(STATE / 'meter.json', {'offset': 0, 'jobs_completed': 0, 'jobs_failed': 0,
|
|
'transfer_failures': 0, 'active': {}}) or {}
|
|
|
|
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(encoding='utf8') 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()
|
|
try:
|
|
(STATE / 'meter.json').write_text(json.dumps(self.data), encoding='utf8')
|
|
except OSError:
|
|
pass
|
|
|
|
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) and read(path).get('uuid') == e['uuid']:
|
|
try:
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
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:
|
|
"""Decision gates: PAUSED, GPU_UNAVAILABLE, DRAINING (power), LOW_DISK,
|
|
THERMAL_THROTTLED, OFFLINE, NETWORK_DEGRADED, BUSY, AVAILABLE.
|
|
Hysteresis and thresholds identical to the Linux controller."""
|
|
|
|
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'] is None or 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'
|
|
|
|
|
|
def status():
|
|
"""Read-only status document (same schema as the Linux worker's status JSON;
|
|
names preserved so central fleet tooling needs no OS branches)."""
|
|
cfg = config()
|
|
now = time.time()
|
|
meter = Meter()
|
|
meter.update()
|
|
admission = Admission(cfg)
|
|
sensors_data = sensors()
|
|
hw = read(STATE / 'hardware.json', {}) or {}
|
|
gpu = hw.get('gpu_model') or gpu_identity()
|
|
proof = read(STATE / 'qsv.json', {}) or {}
|
|
proof_current = bool(proof.get('qsv_h264_ok')) and proof.get('tested_at', 0) > now - 7 * 86400 and \
|
|
(proof.get('boot_id') == hw.get('boot_id') and proof.get('driver_version') == hw.get('driver_version'))
|
|
control = read(STATE / 'control.json', {}) or {}
|
|
active = {}
|
|
try:
|
|
active = ipc('list-jobs', cfg.get('node_binary', 'node'))
|
|
except (RuntimeError, ValueError, OSError):
|
|
active = {}
|
|
sample = {**network(cfg), 'registered': registered(cfg),
|
|
'gpu_accessible': bool(proof_current), 'gpu_ready': bool(proof_current),
|
|
'gpu_tested_at': proof.get('tested_at', 0),
|
|
'on_ac': sensors_data['on_ac'], 'disk_free_bytes': disk_free(),
|
|
'temperature_c': sensors_data['temperature_c'],
|
|
'throttle_counts': sensors_data['throttle_counts'],
|
|
'active_jobs': active.get('active_jobs'), 'concurrency': active.get('concurrency')}
|
|
state = admission.decide(sample, meter, control, now)
|
|
rates = meter.rates(now, cfg.get('bandwidth_ttl_seconds', 900))
|
|
last = meter.data.get('last_encode') or {}
|
|
return {'hostname': os.environ.get('COMPUTERNAME', 'unknown'), 'state': state,
|
|
'tailscale_ip': sample['tailscale_ip'], 'tailscale_path': sample['tailscale_path'],
|
|
'registration': sample['registered'], 'cpu_model': hw.get('cpu_model'),
|
|
'gpu_model': gpu, 'gpu_driver_version': hw.get('driver_version'),
|
|
'qsv_h264_ok': proof.get('qsv_h264_ok'), 'qsv_hevc_ok': proof.get('qsv_hevc_ok'),
|
|
'av1_decode_ok': proof.get('av1_decode_ok'), 'av1_encode_ok': proof.get('av1_encode_ok'),
|
|
'active_jobs': sample['active_jobs'], 'concurrency': sample['concurrency'],
|
|
'gpu_util': last.get('engine_busy_max'), 'temps': sensors_data['temperatures_c'],
|
|
'disk_free': sample['disk_free_bytes'],
|
|
'ram_total_bytes': sensors_data['ram_total_bytes'],
|
|
'ram_available_bytes': sensors_data['ram_available_bytes'],
|
|
'on_ac': sample['on_ac'], 'battery': sensors_data['battery'],
|
|
'recent_download_mbps': rates['recent_download_mbps'],
|
|
'recent_upload_mbps': rates['recent_upload_mbps'],
|
|
'jobs_completed': meter.data.get('jobs_completed', 0),
|
|
'jobs_failed': meter.data.get('jobs_failed', 0),
|
|
'last_job_duration': meter.data.get('last_job_duration_seconds'),
|
|
'last_job_rtf': meter.data.get('last_job_realtime_factor'),
|
|
'tested_at': proof.get('tested_at'), 'boot_id': hw.get('boot_id'),
|
|
'driver_version': hw.get('driver_version')}
|
|
|
|
|
|
def self_test_marker_stale(hw, proof):
|
|
"""Re-proof triggers: boot change, GPU driver change, or 7-day proof age.
|
|
(Extension of the Linux boot-id rule: Windows Update changes graphics
|
|
drivers without a boot.)"""
|
|
if not proof.get('tested_at') or time.time() - proof['tested_at'] > 7 * 86400:
|
|
return True
|
|
return proof.get('boot_id') != hw.get('boot_id') or proof.get('driver_version') != hw.get('driver_version')
|
|
|
|
|
|
def daemon():
|
|
raise SystemExit('daemon loop is a validation item; run status/network-test here, '
|
|
'see windows-worker-result.md section 6 before production enablement')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('command', choices=['status', 'network-test', 'drain'])
|
|
args = parser.parse_args()
|
|
if args.command == 'status':
|
|
print(json.dumps(status(), indent=1))
|
|
elif args.command == 'network-test':
|
|
cfg = config()
|
|
rates = transfer_probe(cfg)
|
|
print(json.dumps(rates, indent=1))
|
|
elif args.command == 'drain':
|
|
cfg = config()
|
|
print(json.dumps(ipc('graceful-shutdown', cfg.get('node_binary', 'node'))))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|