sashatrask
1244 lines
50 KiB
Python
1244 lines
50 KiB
Python
import argparse
|
|
import json
|
|
import os
|
|
import socket
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import yaml
|
|
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
from urllib.parse import urlsplit
|
|
|
|
sys.dont_write_bytecode = True
|
|
if not sys.dont_write_bytecode:
|
|
raise RuntimeError('worker CLI could not disable bytecode writes')
|
|
|
|
from remote_worker_client import WorkerHTTPClient, default_worker_paths
|
|
from runtime_security import (
|
|
atomic_write_private_json,
|
|
canonical_path,
|
|
durable_unlink,
|
|
ensure_private_directory,
|
|
private_file_ready,
|
|
read_private_json,
|
|
)
|
|
from worker_contracts import WORKER_EVENT_SCHEMA
|
|
from worker_local_state import (
|
|
DEFAULT_LOG_BYTES,
|
|
DEFAULT_LOG_FILES,
|
|
DEFAULT_RETENTION_BYTES,
|
|
DEFAULT_RETENTION_DAYS,
|
|
RETENTION_SCHEMA,
|
|
WorkerLocalState,
|
|
WorkerLocalStateError,
|
|
)
|
|
from worker_package import verify_worker_package, worker_package_manifest_sha256
|
|
from worker_supervisor import (
|
|
WorkerAlreadyRunning,
|
|
WorkerSupervisor,
|
|
WorkerSupervisorError,
|
|
CONTROL_SCHEMA,
|
|
PROJECTION_SCHEMA,
|
|
capture_spawned_process_identity,
|
|
classify_instance,
|
|
detached_command,
|
|
load_shutdown_receipt,
|
|
send_control_request,
|
|
spawn_detached,
|
|
terminate_spawned_process,
|
|
wait_for_startup,
|
|
)
|
|
from worker_assignment_runner import run_assignment
|
|
|
|
|
|
CLI_SCHEMA = 1
|
|
CONFIG_SCHEMA = 2
|
|
LEGACY_CONFIG_SCHEMA = 1
|
|
LAUNCH_SCHEMA = 1
|
|
MAX_INSTALL_CONFIG_BYTES = 64 * 1024
|
|
EXIT_SUCCESS = 0
|
|
EXIT_NOT_RUNNING = 3
|
|
EXIT_STALE_OR_UNVERIFIABLE = 4
|
|
EXIT_STARTUP_FAILED = 6
|
|
EXIT_STOP_INCOMPLETE = 7
|
|
EXIT_INVALID_INVOCATION = 64
|
|
|
|
|
|
class InvocationError(ValueError):
|
|
pass
|
|
|
|
|
|
class WorkerArgumentParser(argparse.ArgumentParser):
|
|
def error(self, message):
|
|
raise InvocationError(message)
|
|
|
|
|
|
def _canonical(value):
|
|
return json.dumps(
|
|
value, ensure_ascii=True, sort_keys=True, separators=(',', ':'),
|
|
allow_nan=False,
|
|
)
|
|
|
|
|
|
def _print_json(value):
|
|
print(_canonical(value), flush=True)
|
|
|
|
|
|
def _config_path(paths):
|
|
return os.path.join(paths['state_dir'], 'control', 'worker.config.json')
|
|
|
|
|
|
def _validate_config(value):
|
|
legacy_fields = {
|
|
'schema', 'server', 'token', 'parallelism', 'installed_at',
|
|
}
|
|
current_fields = {
|
|
'schema', 'server', 'token', 'parallelism', 'installed_at',
|
|
'retention_days', 'retention_bytes', 'log_bytes', 'log_files',
|
|
}
|
|
if not isinstance(value, dict):
|
|
raise InvocationError('installed worker configuration is invalid')
|
|
if value.get('schema') == LEGACY_CONFIG_SCHEMA:
|
|
if set(value) == legacy_fields:
|
|
value = {
|
|
**value,
|
|
'schema': CONFIG_SCHEMA,
|
|
'retention_days': DEFAULT_RETENTION_DAYS,
|
|
'retention_bytes': DEFAULT_RETENTION_BYTES,
|
|
'log_bytes': DEFAULT_LOG_BYTES,
|
|
'log_files': DEFAULT_LOG_FILES,
|
|
}
|
|
elif set(value) == current_fields:
|
|
value = {**value, 'schema': CONFIG_SCHEMA}
|
|
else:
|
|
raise InvocationError('installed worker configuration is invalid')
|
|
elif value.get('schema') == CONFIG_SCHEMA and set(value) == current_fields:
|
|
value = dict(value)
|
|
else:
|
|
raise InvocationError('installed worker configuration is invalid')
|
|
if type(value.get('parallelism')) is not int or not 1 <= value['parallelism'] <= 128:
|
|
raise InvocationError('installed worker parallelism is invalid')
|
|
WorkerHTTPClient(value.get('server'), value.get('token'))
|
|
if not isinstance(value.get('installed_at'), str):
|
|
raise InvocationError('installed worker timestamp is invalid')
|
|
if type(value.get('retention_days')) is not int or not 1 <= value['retention_days'] <= 3650:
|
|
raise InvocationError('installed worker retention days are invalid')
|
|
if type(value.get('retention_bytes')) is not int or not 16 * 1024 * 1024 <= value['retention_bytes'] <= 1024 ** 4:
|
|
raise InvocationError('installed worker retention bytes are invalid')
|
|
if type(value.get('log_bytes')) is not int or not 64 * 1024 <= value['log_bytes'] <= 1024 ** 3:
|
|
raise InvocationError('installed worker log bytes are invalid')
|
|
if type(value.get('log_files')) is not int or not 1 <= value['log_files'] <= 20:
|
|
raise InvocationError('installed worker log file count is invalid')
|
|
return dict(value)
|
|
|
|
|
|
def _load_config(paths):
|
|
try:
|
|
return _validate_config(read_private_json(_config_path(paths)))
|
|
except OSError as exc:
|
|
raise InvocationError('worker is not installed; run truf-worker install') from exc
|
|
|
|
|
|
def _load_install_config(path):
|
|
if path == '-':
|
|
payload = sys.stdin.buffer.read(MAX_INSTALL_CONFIG_BYTES + 1)
|
|
else:
|
|
path = os.path.abspath(os.fspath(path))
|
|
if not private_file_ready(path):
|
|
raise InvocationError('worker install configuration must be a private regular file')
|
|
with open(path, 'rb') as handle:
|
|
payload = handle.read(MAX_INSTALL_CONFIG_BYTES + 1)
|
|
if len(payload) > MAX_INSTALL_CONFIG_BYTES:
|
|
raise InvocationError('worker install configuration exceeds its byte bound')
|
|
try:
|
|
class UniqueKeyLoader(yaml.SafeLoader):
|
|
def compose_node(self, parent, index):
|
|
if self.check_event(yaml.AliasEvent):
|
|
event = self.get_event()
|
|
raise yaml.composer.ComposerError(
|
|
None, None, 'YAML aliases are not supported', event.start_mark,
|
|
)
|
|
return super().compose_node(parent, index)
|
|
|
|
def construct_mapping(loader, node, deep=False):
|
|
loader.flatten_mapping(node)
|
|
value = {}
|
|
for key_node, value_node in node.value:
|
|
key = loader.construct_object(key_node, deep=deep)
|
|
if key in value:
|
|
raise yaml.constructor.ConstructorError(
|
|
'while constructing a mapping', node.start_mark,
|
|
f'duplicate key: {key}', key_node.start_mark,
|
|
)
|
|
value[key] = loader.construct_object(value_node, deep=deep)
|
|
return value
|
|
|
|
UniqueKeyLoader.add_constructor(
|
|
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_mapping,
|
|
)
|
|
value = yaml.load(payload.decode('utf-8', errors='strict'), Loader=UniqueKeyLoader)
|
|
except (UnicodeDecodeError, yaml.YAMLError, TypeError, ValueError) as exc:
|
|
raise InvocationError('worker install configuration is invalid YAML') from exc
|
|
if not isinstance(value, dict) or set(value) != {'server', 'token', 'parallelism'}:
|
|
raise InvocationError(
|
|
'worker install configuration must contain only server, token, and parallelism'
|
|
)
|
|
return value
|
|
|
|
|
|
def _configuration(args, paths, *, require_explicit=False):
|
|
server = getattr(args, 'server', None)
|
|
token = getattr(args, 'token', None)
|
|
parallelism = getattr(args, 'parallelism', None)
|
|
config_path = getattr(args, 'config', None)
|
|
if config_path and any(value is not None for value in (server, token, parallelism)):
|
|
raise InvocationError('--config cannot be combined with --server, --token, or --parallelism')
|
|
if config_path:
|
|
document = _load_install_config(config_path)
|
|
server = document['server']
|
|
token = document['token']
|
|
parallelism = document['parallelism']
|
|
if bool(server) != bool(token):
|
|
raise InvocationError('--server and --token must be supplied together')
|
|
if server:
|
|
value = {
|
|
'schema': CONFIG_SCHEMA,
|
|
'server': server,
|
|
'token': token,
|
|
'parallelism': 1 if parallelism is None else parallelism,
|
|
'installed_at': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'),
|
|
'retention_days': int(getattr(args, 'retention_days', DEFAULT_RETENTION_DAYS)),
|
|
'retention_bytes': int(getattr(args, 'retention_bytes', DEFAULT_RETENTION_BYTES)),
|
|
'log_bytes': int(getattr(args, 'log_bytes', DEFAULT_LOG_BYTES)),
|
|
'log_files': int(getattr(args, 'log_files', DEFAULT_LOG_FILES)),
|
|
}
|
|
return _validate_config(value)
|
|
if require_explicit:
|
|
raise InvocationError('--server and --token are required')
|
|
value = _load_config(paths)
|
|
if parallelism is not None:
|
|
value['parallelism'] = parallelism
|
|
return _validate_config(value)
|
|
|
|
|
|
def _client_args(config, paths):
|
|
return SimpleNamespace(
|
|
package_manifest=paths['package_manifest'],
|
|
state_dir=paths['state_dir'],
|
|
bundle_dir=paths['bundle_dir'],
|
|
work_dir=paths['work_dir'],
|
|
server=config['server'],
|
|
token=config['token'],
|
|
parallelism=config['parallelism'],
|
|
poll_seconds=5.0,
|
|
error_delay_seconds=15.0,
|
|
http_timeout=120,
|
|
retention_days=config['retention_days'],
|
|
retention_bytes=config['retention_bytes'],
|
|
log_bytes=config['log_bytes'],
|
|
log_files=config['log_files'],
|
|
)
|
|
|
|
|
|
def _add_configuration(parser, *, required=False, allow_config=False):
|
|
parser.add_argument('--server', required=required)
|
|
parser.add_argument('--token', required=required)
|
|
parser.add_argument('--parallelism', type=int)
|
|
if allow_config:
|
|
parser.add_argument('--config')
|
|
|
|
|
|
def build_parser():
|
|
parser = WorkerArgumentParser(prog='truf-worker', description='TRUF remote worker operator CLI')
|
|
commands = parser.add_subparsers(dest='command', required=True)
|
|
|
|
install = commands.add_parser('install')
|
|
_add_configuration(install, allow_config=True)
|
|
install.add_argument('--retention-days', type=int, default=DEFAULT_RETENTION_DAYS)
|
|
install.add_argument('--retention-bytes', type=int, default=DEFAULT_RETENTION_BYTES)
|
|
install.add_argument('--log-bytes', type=int, default=DEFAULT_LOG_BYTES)
|
|
install.add_argument('--log-files', type=int, default=DEFAULT_LOG_FILES)
|
|
run = commands.add_parser('run')
|
|
_add_configuration(run)
|
|
start = commands.add_parser('start')
|
|
_add_configuration(start)
|
|
start.add_argument('--startup-timeout', type=float, default=30.0)
|
|
|
|
stop = commands.add_parser('stop')
|
|
stop.add_argument('--timeout', type=float, default=30.0)
|
|
stop.add_argument('--json', action='store_true')
|
|
|
|
status = commands.add_parser('status')
|
|
status.add_argument('--json', action='store_true')
|
|
|
|
attach = commands.add_parser('attach')
|
|
attach_output = attach.add_mutually_exclusive_group()
|
|
attach_output.add_argument('--json', action='store_true')
|
|
attach_output.add_argument('--ndjson', action='store_true')
|
|
attach.add_argument('--follow-seconds', type=float)
|
|
|
|
watch = commands.add_parser('watch')
|
|
watch.add_argument('--follow-seconds', type=float)
|
|
watch.set_defaults(json=False, ndjson=False)
|
|
|
|
logs = commands.add_parser('logs')
|
|
logs.add_argument('--tail', type=int, default=100)
|
|
logs.add_argument('--follow', action='store_true')
|
|
logs_output = logs.add_mutually_exclusive_group()
|
|
logs_output.add_argument('--json', action='store_true')
|
|
logs_output.add_argument('--ndjson', action='store_true')
|
|
logs.add_argument('--follow-seconds', type=float, default=60.0)
|
|
|
|
history = commands.add_parser('history')
|
|
history.add_argument('--limit', type=int, default=100)
|
|
history.add_argument('--reservation', type=int)
|
|
history_output = history.add_mutually_exclusive_group()
|
|
history_output.add_argument('--json', action='store_true')
|
|
history_output.add_argument('--ndjson', action='store_true')
|
|
|
|
doctor = commands.add_parser('doctor')
|
|
doctor.add_argument('--json', action='store_true')
|
|
|
|
child = commands.add_parser('_supervise', help=argparse.SUPPRESS)
|
|
child.add_argument('--launch-file', required=True)
|
|
child.add_argument('--startup-file', required=True)
|
|
child.add_argument('--launch-nonce', required=True)
|
|
runner = commands.add_parser('_assignment_runner', help=argparse.SUPPRESS)
|
|
runner.add_argument('--root', required=True)
|
|
return parser
|
|
|
|
|
|
def parse_args(argv=None):
|
|
values = list(sys.argv[1:] if argv is None else argv)
|
|
if values and values[0].startswith('-'):
|
|
values.insert(0, 'run')
|
|
args = build_parser().parse_args(values)
|
|
if getattr(args, 'parallelism', None) is not None and not 1 <= args.parallelism <= 128:
|
|
raise InvocationError('--parallelism must be between 1 and 128')
|
|
if getattr(args, 'startup_timeout', 1) is not None and not 0.1 <= getattr(args, 'startup_timeout', 1) <= 300:
|
|
raise InvocationError('--startup-timeout must be between 0.1 and 300 seconds')
|
|
if getattr(args, 'timeout', 1) is not None and not 0.1 <= getattr(args, 'timeout', 1) <= 3600:
|
|
raise InvocationError('--timeout must be between 0.1 and 3600 seconds')
|
|
if getattr(args, 'tail', 1) is not None and not 1 <= getattr(args, 'tail', 1) <= 10000:
|
|
raise InvocationError('--tail must be between 1 and 10000')
|
|
if getattr(args, 'limit', 1) is not None and not 1 <= getattr(args, 'limit', 1) <= 1000:
|
|
raise InvocationError('--limit must be between 1 and 1000')
|
|
follow = getattr(args, 'follow_seconds', None)
|
|
if follow is not None and not 0 <= follow <= 3600:
|
|
raise InvocationError('--follow-seconds must be between 0 and 3600')
|
|
if getattr(args, 'reservation', None) is not None and args.reservation <= 0:
|
|
raise InvocationError('--reservation must be positive')
|
|
return args
|
|
|
|
|
|
def _seconds(value, now=None):
|
|
if not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
|
|
except ValueError:
|
|
return None
|
|
return round(((now or datetime.now(timezone.utc)) - parsed).total_seconds(), 3)
|
|
|
|
|
|
def _local_layout_ready(state_dir):
|
|
return all(os.path.isdir(os.path.join(state_dir, name)) for name in (
|
|
'control', 'events', 'history', 'diagnostics', 'logs',
|
|
))
|
|
|
|
|
|
def _local_settings(paths):
|
|
try:
|
|
config = _load_config(paths)
|
|
except (InvocationError, ValueError):
|
|
config = {}
|
|
return {
|
|
'retention_days': config.get('retention_days', DEFAULT_RETENTION_DAYS),
|
|
'retention_bytes': config.get('retention_bytes', DEFAULT_RETENTION_BYTES),
|
|
'log_bytes': config.get('log_bytes', DEFAULT_LOG_BYTES),
|
|
'log_files': config.get('log_files', DEFAULT_LOG_FILES),
|
|
}
|
|
|
|
|
|
def _open_local_reader(paths):
|
|
return WorkerLocalState(
|
|
paths['state_dir'], read_only=True, **_local_settings(paths),
|
|
)
|
|
|
|
|
|
def _directory_usage(path):
|
|
byte_count = 0
|
|
file_count = 0
|
|
if os.path.isdir(path):
|
|
for current, directories, files in os.walk(path, followlinks=False):
|
|
directories[:] = [
|
|
name for name in directories
|
|
if not os.path.islink(os.path.join(current, name))
|
|
]
|
|
for name in files:
|
|
candidate = os.path.join(current, name)
|
|
try:
|
|
if os.path.isfile(candidate) and not os.path.islink(candidate):
|
|
byte_count += os.path.getsize(candidate)
|
|
file_count += 1
|
|
except OSError:
|
|
continue
|
|
return {'bytes': byte_count, 'files': file_count}
|
|
|
|
|
|
def _root_file_usage(path):
|
|
byte_count = 0
|
|
file_count = 0
|
|
if os.path.isdir(path):
|
|
with os.scandir(path) as entries:
|
|
for entry in entries:
|
|
try:
|
|
if entry.is_file(follow_symlinks=False):
|
|
byte_count += entry.stat(follow_symlinks=False).st_size
|
|
file_count += 1
|
|
except OSError:
|
|
continue
|
|
return {'bytes': byte_count, 'files': file_count}
|
|
|
|
|
|
def _fallback_retention(paths):
|
|
settings = _local_settings(paths)
|
|
def non_evictable(item):
|
|
return {
|
|
**item,
|
|
'evictable_bytes': 0,
|
|
'evictable_files': 0,
|
|
'non_evictable_bytes': item['bytes'],
|
|
'non_evictable_files': item['files'],
|
|
}
|
|
categories = {
|
|
'control': non_evictable(_directory_usage(
|
|
os.path.join(paths['state_dir'], 'control'),
|
|
)),
|
|
'state': non_evictable(_root_file_usage(paths['state_dir'])),
|
|
'events': non_evictable({'bytes': 0, 'files': 0}),
|
|
'history': non_evictable({'bytes': 0, 'files': 0}),
|
|
'diagnostics': non_evictable({'bytes': 0, 'files': 0}),
|
|
'logs': non_evictable({'bytes': 0, 'files': 0}),
|
|
'bundles': non_evictable(_directory_usage(paths['bundle_dir'])),
|
|
'work': non_evictable(_directory_usage(paths['work_dir'])),
|
|
}
|
|
total_bytes = sum(item['bytes'] for item in categories.values())
|
|
return {
|
|
'schema': RETENTION_SCHEMA,
|
|
'maximum_age_days': settings['retention_days'],
|
|
'maximum_bytes': settings['retention_bytes'],
|
|
'total_bytes': total_bytes,
|
|
'total_files': sum(item['files'] for item in categories.values()),
|
|
'evictable_bytes': 0,
|
|
'evictable_files': 0,
|
|
'non_evictable_bytes': total_bytes,
|
|
'non_evictable_files': sum(item['files'] for item in categories.values()),
|
|
'over_limit': total_bytes > settings['retention_bytes'],
|
|
'categories': categories,
|
|
}
|
|
|
|
|
|
def _slot_document(slot, now=None):
|
|
now = now or datetime.now(timezone.utc)
|
|
progress = dict(slot.get('progress') or {})
|
|
assignment_remaining = _seconds(slot.get('assignment_deadline_at'), now)
|
|
scan_remaining = _seconds(slot.get('scan_deadline_at'), now)
|
|
measured = {
|
|
key: value
|
|
for key, value in progress.items()
|
|
if type(value) in (int, float) and key not in {
|
|
'attempt', 'retry_after_seconds',
|
|
}
|
|
}
|
|
counters = progress.get('counters')
|
|
if isinstance(counters, dict):
|
|
measured.update({
|
|
key: value for key, value in counters.items()
|
|
if type(value) in (int, float)
|
|
})
|
|
return {
|
|
'slot_id': int(slot['slot_id']),
|
|
'sequence': int(slot.get('sequence') or 0),
|
|
'phase': str(slot['phase']),
|
|
'phase_age_seconds': _seconds(slot.get('phase_started_at'), now),
|
|
'reservation_id': slot.get('reservation_id'),
|
|
'source': slot.get('source'),
|
|
'attempt': progress.get('attempt'),
|
|
'scan_deadline_at': slot.get('scan_deadline_at'),
|
|
'scan_deadline_remaining_seconds': (
|
|
-scan_remaining if scan_remaining is not None else None
|
|
),
|
|
'assignment_deadline_at': slot.get('assignment_deadline_at'),
|
|
'assignment_remaining_seconds': (
|
|
-assignment_remaining if assignment_remaining is not None else None
|
|
),
|
|
'last_progress_at': slot.get('timestamp'),
|
|
'last_progress_age_seconds': _seconds(slot.get('timestamp'), now),
|
|
'progress': progress,
|
|
'measured_progress': measured,
|
|
'child_state': progress.get('child_state'),
|
|
'idle_reason': progress.get('reason') if slot.get('phase') in {'idle', 'backoff'} else None,
|
|
'next_claim_at': progress.get('next_claim_at'),
|
|
}
|
|
|
|
|
|
def status_document(paths, classification=None):
|
|
classification = classification or classify_instance(paths['state_dir'])
|
|
state = classification['state']
|
|
instance = classification.get('instance')
|
|
package = instance.get('package') if instance else None
|
|
runtime = instance.get('runtime') if instance else None
|
|
protocol = instance.get('protocol') if instance else None
|
|
if instance is None:
|
|
try:
|
|
verified = verify_worker_package(paths['package_manifest'])
|
|
manifest = verified['manifest']
|
|
package = {
|
|
'schema': manifest['schema'],
|
|
'manifest_sha256': worker_package_manifest_sha256(manifest),
|
|
'code_manifest_sha256': verified['code_manifest_sha256'],
|
|
'platform_tag': manifest['platform_tag'],
|
|
}
|
|
runtime = {
|
|
'python': '.'.join(str(item) for item in sys.version_info[:3]),
|
|
'platform': sys.platform,
|
|
'executable': canonical_path(sys.executable),
|
|
'mode': 'inactive',
|
|
}
|
|
protocol = {
|
|
'worker_protocol': manifest['protocol_version'],
|
|
'bundle_format': manifest['bundle_format_version'],
|
|
'event': WORKER_EVENT_SCHEMA,
|
|
'control': CONTROL_SCHEMA,
|
|
'projection': PROJECTION_SCHEMA,
|
|
}
|
|
except (OSError, ValueError):
|
|
pass
|
|
worker = None
|
|
slots = []
|
|
retention = None
|
|
if state in {'running', 'draining'}:
|
|
try:
|
|
snapshot = send_control_request(classification['record'], 'snapshot', {})
|
|
worker = snapshot['worker']
|
|
slots = [_slot_document(item) for item in snapshot['projection']['slots']]
|
|
retention = snapshot['retention']
|
|
except (OSError, ValueError, WorkerSupervisorError, WorkerLocalStateError) as exc:
|
|
state = 'unverifiable'
|
|
classification = dict(classification)
|
|
classification['state'] = state
|
|
classification['detail'] = f'live worker snapshot failed: {type(exc).__name__}'
|
|
if state not in {'running', 'draining'} and os.path.isdir(paths['state_dir']):
|
|
local = None
|
|
if _local_layout_ready(paths['state_dir']):
|
|
try:
|
|
local = _open_local_reader(paths)
|
|
except (OSError, ValueError, WorkerLocalStateError):
|
|
local = None
|
|
projection = (
|
|
local.snapshot()
|
|
if local is not None
|
|
else {'aggregate': {'slot_count': 0, 'phases': {}}, 'slots': []}
|
|
)
|
|
slots = [_slot_document(item) for item in projection['slots']]
|
|
retention = (
|
|
local.retention_usage({
|
|
'bundles': paths['bundle_dir'], 'work': paths['work_dir'],
|
|
}) if local is not None else _fallback_retention(paths)
|
|
)
|
|
try:
|
|
configured_parallelism = _load_config(paths)['parallelism']
|
|
except (InvocationError, ValueError):
|
|
configured_parallelism = None
|
|
worker = {
|
|
'state': state,
|
|
'parallelism': configured_parallelism,
|
|
'slot_cap': configured_parallelism,
|
|
'configured_slots': configured_parallelism,
|
|
'recovery_slots': None,
|
|
'started_at': None,
|
|
'drain_deadline_at': None,
|
|
'aggregate': projection['aggregate'],
|
|
}
|
|
return {
|
|
'schema': CLI_SCHEMA,
|
|
'command': 'status',
|
|
'state': state,
|
|
'detail': classification.get('detail'),
|
|
'instance': instance,
|
|
'package': package,
|
|
'runtime': runtime,
|
|
'protocol': protocol,
|
|
'worker': worker,
|
|
'slots': slots,
|
|
'retention': retention,
|
|
}
|
|
|
|
|
|
def _human_status(document):
|
|
print(f"Worker: {document['state']} ({document.get('detail') or 'no detail'})")
|
|
instance = document.get('instance')
|
|
if instance:
|
|
print(f"Instance: {instance['instance_id']} PID: {instance['pid']} Mode: {instance['runtime']['mode']}")
|
|
print(
|
|
f"Package: {instance['package']['manifest_sha256']} "
|
|
f"Protocol: {instance['protocol']['worker_protocol']}"
|
|
)
|
|
worker = document.get('worker')
|
|
if worker:
|
|
print(
|
|
f"Slots: {worker['aggregate']['slot_count']} Cap: {worker['slot_cap']} "
|
|
f"State: {worker['state']}"
|
|
)
|
|
for slot in document.get('slots') or []:
|
|
reason = slot['idle_reason'] or ''
|
|
visibility = slot['progress'].get('execution_visibility') or ''
|
|
counters = _canonical(slot['measured_progress']) if slot['measured_progress'] else '{}'
|
|
print(
|
|
f"slot {slot['slot_id']}: {slot['phase']} age={slot['phase_age_seconds']}s "
|
|
f"source={slot['source'] or '-'} reservation={slot['reservation_id'] or '-'} "
|
|
f"attempt={slot['attempt'] or '-'} "
|
|
f"scan_deadline={slot['scan_deadline_at'] or '-'} "
|
|
f"scan_remaining={slot['scan_deadline_remaining_seconds']}s "
|
|
f"assignment_deadline={slot['assignment_deadline_at'] or '-'} "
|
|
f"assignment_remaining={slot['assignment_remaining_seconds']}s "
|
|
f"progress_age={slot['last_progress_age_seconds']}s "
|
|
f"next_claim_at={slot['next_claim_at'] or '-'} "
|
|
f"child_state={slot['child_state'] or '-'} counters={counters} "
|
|
f"{reason} {visibility}"
|
|
)
|
|
retention = document.get('retention')
|
|
if retention:
|
|
print(f"Local retention: {retention['total_bytes']} bytes in {retention['total_files']} files")
|
|
|
|
|
|
def command_install(args, paths):
|
|
config = _configuration(args, paths, require_explicit=True)
|
|
try:
|
|
verify_worker_package(paths['package_manifest'])
|
|
except Exception as exc:
|
|
print(f'Worker installation failed: {type(exc).__name__}', file=sys.stderr)
|
|
return EXIT_STARTUP_FAILED
|
|
ensure_private_directory(paths['state_dir'], reject_reparse=True)
|
|
ensure_private_directory(os.path.join(paths['state_dir'], 'control'), reject_reparse=True)
|
|
ensure_private_directory(paths['bundle_dir'], reject_reparse=True)
|
|
ensure_private_directory(paths['work_dir'], reject_reparse=True)
|
|
atomic_write_private_json(_config_path(paths), config)
|
|
print(f"Installed worker configuration at {_config_path(paths)}")
|
|
return EXIT_SUCCESS
|
|
|
|
|
|
def command_run(args, paths):
|
|
config = _configuration(args, paths)
|
|
try:
|
|
supervisor = WorkerSupervisor(_client_args(config, paths), foreground=True)
|
|
return supervisor.run()
|
|
except WorkerAlreadyRunning as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return EXIT_STARTUP_FAILED
|
|
except Exception as exc:
|
|
print(f'Worker startup failed: {type(exc).__name__}', file=sys.stderr)
|
|
return EXIT_STARTUP_FAILED
|
|
|
|
|
|
def _launch_paths(paths, nonce):
|
|
control = ensure_private_directory(os.path.join(paths['state_dir'], 'control'), reject_reparse=True)
|
|
return (
|
|
os.path.join(control, f'worker.launch.{nonce}.json'),
|
|
os.path.join(control, f'worker.startup.{nonce}.json'),
|
|
)
|
|
|
|
|
|
def command_start(args, paths):
|
|
current = classify_instance(paths['state_dir'], remove_stale=True)
|
|
if current['state'] == 'running':
|
|
document = status_document(paths, current)
|
|
_human_status(document)
|
|
return EXIT_SUCCESS
|
|
if current['state'] in {'unverifiable', 'starting', 'draining'} or (
|
|
current['state'] == 'stale' and not current.get('removed')
|
|
):
|
|
_human_status(status_document(paths, current))
|
|
return EXIT_STALE_OR_UNVERIFIABLE
|
|
config = _configuration(args, paths)
|
|
nonce = os.urandom(16).hex()
|
|
launch_file, startup_file = _launch_paths(paths, nonce)
|
|
launch = {
|
|
'schema': LAUNCH_SCHEMA,
|
|
'nonce': nonce,
|
|
'server': config['server'],
|
|
'token': config['token'],
|
|
'parallelism': config['parallelism'],
|
|
'retention_days': config['retention_days'],
|
|
'retention_bytes': config['retention_bytes'],
|
|
'log_bytes': config['log_bytes'],
|
|
'log_files': config['log_files'],
|
|
'package_manifest': paths['package_manifest'],
|
|
'state_dir': paths['state_dir'],
|
|
'bundle_dir': paths['bundle_dir'],
|
|
'work_dir': paths['work_dir'],
|
|
}
|
|
atomic_write_private_json(launch_file, launch)
|
|
bootstrap = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'remote_worker_bootstrap.py')
|
|
process = None
|
|
spawned_identity = None
|
|
started = False
|
|
try:
|
|
process = spawn_detached(detached_command(bootstrap, launch_file, startup_file, nonce))
|
|
spawned_identity = capture_spawned_process_identity(process)
|
|
result = wait_for_startup(startup_file, nonce, process, timeout=args.startup_timeout)
|
|
if result['outcome'] != 'ready':
|
|
print(f"Worker startup failed: {result['outcome']}", file=sys.stderr)
|
|
return EXIT_STARTUP_FAILED
|
|
verified = classify_instance(paths['state_dir'])
|
|
if (
|
|
verified['state'] != 'running'
|
|
or verified['instance']['instance_id'] != result['instance_id']
|
|
or result['pid'] != spawned_identity['pid']
|
|
or verified['instance']['pid'] != spawned_identity['pid']
|
|
or verified['instance']['process_creation_time'] != spawned_identity['creation_time']
|
|
or canonical_path(verified['instance']['executable']) != canonical_path(
|
|
spawned_identity['executable']
|
|
)
|
|
):
|
|
print('Worker startup identity could not be verified', file=sys.stderr)
|
|
return EXIT_STARTUP_FAILED
|
|
started = True
|
|
_human_status(status_document(paths, verified))
|
|
return EXIT_SUCCESS
|
|
except (OSError, ValueError, WorkerSupervisorError) as exc:
|
|
print(f'Worker startup failed: {exc}', file=sys.stderr)
|
|
return EXIT_STARTUP_FAILED
|
|
finally:
|
|
try:
|
|
if process is not None and not started:
|
|
if spawned_identity is not None:
|
|
terminate_spawned_process(process, spawned_identity)
|
|
elif process.poll() is None:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
classify_instance(paths['state_dir'], remove_stale=True)
|
|
finally:
|
|
for path in (launch_file, startup_file):
|
|
try:
|
|
if os.path.exists(path):
|
|
durable_unlink(path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def command_stop(args, paths):
|
|
classification = classify_instance(paths['state_dir'])
|
|
if classification['state'] == 'stopped':
|
|
value = {'schema': CLI_SCHEMA, 'command': 'stop', 'state': 'stopped', 'receipt': None}
|
|
_print_json(value) if args.json else print('Worker is not running')
|
|
return EXIT_NOT_RUNNING
|
|
if classification['state'] not in {'running', 'draining'}:
|
|
value = {
|
|
'schema': CLI_SCHEMA, 'command': 'stop',
|
|
'state': classification['state'], 'receipt': None,
|
|
}
|
|
_print_json(value) if args.json else print(f"Worker is {classification['state']}")
|
|
return EXIT_STALE_OR_UNVERIFIABLE
|
|
record = classification['record']
|
|
try:
|
|
result = send_control_request(
|
|
record, 'stop', {'timeout_seconds': args.timeout}, timeout=min(args.timeout, 10),
|
|
)
|
|
except (OSError, TimeoutError, WorkerSupervisorError):
|
|
final = status_document(paths, classify_instance(paths['state_dir']))
|
|
value = {
|
|
'schema': CLI_SCHEMA, 'command': 'stop',
|
|
'state': 'control_disconnected', 'receipt': None,
|
|
'final_status': final,
|
|
}
|
|
if args.json:
|
|
_print_json(value)
|
|
else:
|
|
print('Worker control disconnected; final state:')
|
|
_human_status(final)
|
|
return (
|
|
EXIT_NOT_RUNNING if final['state'] == 'stopped'
|
|
else EXIT_STALE_OR_UNVERIFIABLE
|
|
)
|
|
deadline = time.monotonic() + args.timeout + 5.0
|
|
receipt = None
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
receipt = load_shutdown_receipt(paths['state_dir'], record['instance_id'])
|
|
break
|
|
except (OSError, WorkerSupervisorError):
|
|
time.sleep(0.1)
|
|
clean = bool(
|
|
receipt is not None
|
|
and receipt.get('drained') is True
|
|
and receipt.get('exit_code') == 0
|
|
)
|
|
state = 'stopped' if clean else (
|
|
'non_drained' if receipt is not None else 'stop_timeout'
|
|
)
|
|
value = {
|
|
'schema': CLI_SCHEMA, 'command': 'stop', 'state': state,
|
|
'accepted': result['accepted'], 'receipt': receipt,
|
|
}
|
|
_print_json(value) if args.json else print(
|
|
'Worker stopped cleanly' if clean else (
|
|
'Worker exited without a clean drain receipt'
|
|
if receipt is not None else 'Worker stop timed out without a shutdown receipt'
|
|
)
|
|
)
|
|
return EXIT_SUCCESS if clean else EXIT_STOP_INCOMPLETE
|
|
|
|
|
|
def command_status(args, paths):
|
|
document = status_document(paths)
|
|
_print_json(document) if args.json else _human_status(document)
|
|
if document['state'] in {'running', 'starting', 'draining'}:
|
|
return EXIT_SUCCESS
|
|
if document['state'] == 'stopped':
|
|
return EXIT_NOT_RUNNING
|
|
return EXIT_STALE_OR_UNVERIFIABLE
|
|
|
|
|
|
def _follow_events(record, sequence, seconds, emit):
|
|
deadline = None if seconds is None else time.monotonic() + seconds
|
|
while deadline is None or time.monotonic() < deadline:
|
|
response = send_control_request(
|
|
record, 'events', {'after_sequence': sequence, 'limit': 256},
|
|
)
|
|
for event in response['events']:
|
|
sequence = max(sequence, int(event['sequence']))
|
|
emit(event)
|
|
time.sleep(0.2)
|
|
return sequence
|
|
|
|
|
|
def command_attach(args, paths):
|
|
command_name = getattr(args, 'command', 'attach')
|
|
classification = classify_instance(paths['state_dir'])
|
|
if classification['state'] not in {'running', 'draining'}:
|
|
document = status_document(paths, classification)
|
|
if args.json:
|
|
_print_json({'schema': CLI_SCHEMA, 'command': command_name, 'status': document})
|
|
elif args.ndjson:
|
|
_print_json({'schema': CLI_SCHEMA, 'type': 'snapshot', 'status': document})
|
|
else:
|
|
_human_status(document)
|
|
return (
|
|
EXIT_NOT_RUNNING if classification['state'] == 'stopped'
|
|
else EXIT_STALE_OR_UNVERIFIABLE
|
|
)
|
|
record = classification['record']
|
|
snapshot = status_document(paths, classification)
|
|
if args.json:
|
|
_print_json({
|
|
'schema': CLI_SCHEMA,
|
|
'command': command_name,
|
|
'status': snapshot,
|
|
})
|
|
return EXIT_SUCCESS
|
|
if args.ndjson:
|
|
_print_json({'schema': CLI_SCHEMA, 'type': 'snapshot', 'status': snapshot})
|
|
else:
|
|
_human_status(snapshot)
|
|
action = 'Watching' if command_name == 'watch' else 'Attached'
|
|
print(f'{action}; q, EOF, or Ctrl-C detaches without stopping the worker.')
|
|
stop = threading.Event()
|
|
if not args.ndjson:
|
|
def read_input():
|
|
try:
|
|
while not stop.is_set():
|
|
value = sys.stdin.read(1)
|
|
if value == '' or value.lower() == 'q':
|
|
stop.set()
|
|
return
|
|
except (EOFError, OSError):
|
|
stop.set()
|
|
threading.Thread(target=read_input, daemon=True).start()
|
|
sequence = max((slot.get('sequence', 0) for slot in snapshot.get('slots', [])), default=0)
|
|
deadline = None if args.follow_seconds is None else time.monotonic() + args.follow_seconds
|
|
next_refresh = time.monotonic() + 1.0
|
|
disconnected = False
|
|
try:
|
|
while not stop.is_set() and (deadline is None or time.monotonic() < deadline):
|
|
response = send_control_request(
|
|
record, 'events', {'after_sequence': sequence, 'limit': 256},
|
|
)
|
|
for event in response['events']:
|
|
sequence = max(sequence, int(event['sequence']))
|
|
if args.ndjson:
|
|
_print_json(event)
|
|
now = time.monotonic()
|
|
if not args.ndjson and (response['events'] or now >= next_refresh):
|
|
refreshed = status_document(paths, classification)
|
|
print('---')
|
|
_human_status(refreshed)
|
|
next_refresh = now + 1.0
|
|
stop.wait(0.2)
|
|
except (OSError, TimeoutError, WorkerSupervisorError):
|
|
disconnected = True
|
|
final = status_document(paths, classify_instance(paths['state_dir']))
|
|
if args.ndjson:
|
|
_print_json({'schema': CLI_SCHEMA, 'type': 'snapshot', 'status': final})
|
|
else:
|
|
print('Worker control disconnected; final state:')
|
|
_human_status(final)
|
|
except (KeyboardInterrupt, EOFError):
|
|
pass
|
|
finally:
|
|
stop.set()
|
|
if disconnected:
|
|
return (
|
|
EXIT_NOT_RUNNING if final['state'] == 'stopped'
|
|
else EXIT_STALE_OR_UNVERIFIABLE
|
|
)
|
|
return EXIT_SUCCESS
|
|
|
|
|
|
def command_logs(args, paths):
|
|
if not os.path.isdir(paths['state_dir']):
|
|
value = {'schema': CLI_SCHEMA, 'command': 'logs', 'lines': []}
|
|
if args.follow and (args.json or args.ndjson):
|
|
print('Worker logs follow unavailable: worker state is not initialized', file=sys.stderr)
|
|
elif args.json:
|
|
_print_json(value)
|
|
elif not args.ndjson:
|
|
print('No worker logs')
|
|
return EXIT_NOT_RUNNING
|
|
layout_ready = _local_layout_ready(paths['state_dir'])
|
|
local = _open_local_reader(paths) if layout_ready else None
|
|
lines = local.log_tail(args.tail) if local is not None else []
|
|
machine_follow = args.follow and (args.json or args.ndjson)
|
|
if machine_follow:
|
|
sequence = max(0, (local.snapshot()['sequence'] if local else 0) - args.tail)
|
|
events = local.events_after(sequence, args.tail) if local else []
|
|
for event in events:
|
|
_print_json(event)
|
|
if events:
|
|
sequence = events[-1]['sequence']
|
|
elif args.ndjson:
|
|
for index, line in enumerate(lines, 1):
|
|
_print_json({'schema': CLI_SCHEMA, 'type': 'log', 'index': index, 'line': line})
|
|
elif args.json and not args.follow:
|
|
_print_json({'schema': CLI_SCHEMA, 'command': 'logs', 'lines': lines})
|
|
else:
|
|
for line in lines:
|
|
print(line)
|
|
if not args.follow:
|
|
return EXIT_SUCCESS
|
|
classification = classify_instance(paths['state_dir'])
|
|
if classification['state'] not in {'running', 'draining'}:
|
|
if machine_follow:
|
|
print(
|
|
f"Worker logs follow ended: {classification['state']}",
|
|
file=sys.stderr,
|
|
)
|
|
return EXIT_NOT_RUNNING
|
|
if local is None:
|
|
local = _open_local_reader(paths)
|
|
if not machine_follow:
|
|
sequence = local.snapshot()['sequence']
|
|
try:
|
|
_follow_events(
|
|
classification['record'], sequence, args.follow_seconds,
|
|
_print_json if machine_follow else lambda event: print(
|
|
f"#{event['sequence']} slot {event['slot_id']} {event['phase']}"
|
|
),
|
|
)
|
|
except (OSError, TimeoutError, WorkerSupervisorError):
|
|
final = status_document(paths, classify_instance(paths['state_dir']))
|
|
if machine_follow:
|
|
print(
|
|
f"Worker logs follow control disconnected: {final['state']}",
|
|
file=sys.stderr,
|
|
)
|
|
else:
|
|
print('Worker control disconnected; final state:')
|
|
_human_status(final)
|
|
return (
|
|
EXIT_NOT_RUNNING if final['state'] == 'stopped'
|
|
else EXIT_STALE_OR_UNVERIFIABLE
|
|
)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
return EXIT_SUCCESS
|
|
|
|
|
|
def command_history(args, paths):
|
|
if not _local_layout_ready(paths['state_dir']):
|
|
values = []
|
|
else:
|
|
values = _open_local_reader(paths).history(args.limit, args.reservation)
|
|
if args.ndjson:
|
|
for value in values:
|
|
_print_json(value)
|
|
elif args.json:
|
|
_print_json({'schema': CLI_SCHEMA, 'command': 'history', 'assignments': values})
|
|
elif not values:
|
|
print('No terminal assignment history')
|
|
else:
|
|
for value in values:
|
|
print(
|
|
f"reservation {value['reservation_id']}: {value['outcome']} "
|
|
f"source={value.get('source') or '-'} slot={value['slot_id']} "
|
|
f"completed={value['completed_at']} duration={value.get('duration_seconds')}s"
|
|
)
|
|
timeline = ' -> '.join(
|
|
f"{event['phase']}#{event['sequence']}"
|
|
for event in value.get('timeline') or []
|
|
)
|
|
if timeline:
|
|
print(f' timeline {timeline}')
|
|
for diagnostic in value.get('diagnostics') or []:
|
|
print(
|
|
f" diagnostic {diagnostic.get('diagnostic_uid')}: "
|
|
f"{diagnostic.get('record')} available={diagnostic.get('available')}"
|
|
)
|
|
return EXIT_SUCCESS
|
|
|
|
|
|
def _doctor_check(name, applicable, status, summary):
|
|
return {
|
|
'name': name,
|
|
'applicable': bool(applicable),
|
|
'status': str(status),
|
|
'summary': str(summary),
|
|
}
|
|
|
|
|
|
def command_doctor(args, paths):
|
|
checks = []
|
|
package = None
|
|
installed = os.path.isfile(paths['package_manifest'])
|
|
checks.append(_doctor_check(
|
|
'installation', True, 'ok' if installed else 'error',
|
|
'worker package manifest is installed' if installed else 'worker package manifest is absent',
|
|
))
|
|
try:
|
|
package = verify_worker_package(paths['package_manifest'])
|
|
checks.append(_doctor_check('package_identity', True, 'ok', 'package identity verified'))
|
|
except Exception as exc:
|
|
checks.append(_doctor_check('package_identity', True, 'error', f'package verification failed: {type(exc).__name__}'))
|
|
config = None
|
|
try:
|
|
config = _load_config(paths)
|
|
checks.append(_doctor_check('configuration', True, 'ok', 'installed configuration is valid'))
|
|
checks.append(_doctor_check('credentials', True, 'ok', 'worker credential shape is valid'))
|
|
except (InvocationError, ValueError) as exc:
|
|
checks.append(_doctor_check('configuration', True, 'error', str(exc)))
|
|
checks.append(_doctor_check('credentials', False, 'not_applicable', 'configuration is unavailable'))
|
|
path_values = (paths['state_dir'], paths['bundle_dir'], paths['work_dir'])
|
|
missing_paths = [path for path in path_values if not os.path.isdir(path)]
|
|
if missing_paths:
|
|
checks.append(_doctor_check(
|
|
'paths', False, 'not_applicable',
|
|
'runtime paths are not initialized; run install first',
|
|
))
|
|
else:
|
|
probe_error = None
|
|
for path in path_values:
|
|
descriptor = None
|
|
temporary = None
|
|
try:
|
|
descriptor, temporary = tempfile.mkstemp(prefix='.doctor-', dir=path)
|
|
except OSError as exc:
|
|
probe_error = exc
|
|
break
|
|
finally:
|
|
if descriptor is not None:
|
|
os.close(descriptor)
|
|
if temporary is not None:
|
|
try:
|
|
os.unlink(temporary)
|
|
except FileNotFoundError:
|
|
pass
|
|
except OSError as exc:
|
|
probe_error = probe_error or exc
|
|
if probe_error is None:
|
|
checks.append(_doctor_check('paths', True, 'ok', 'initialized state and data paths are writable'))
|
|
else:
|
|
checks.append(_doctor_check(
|
|
'paths', True, 'error',
|
|
f'path validation failed: {type(probe_error).__name__}',
|
|
))
|
|
for name, key in (('git', 'git_path'), ('trufflehog', 'trufflehog_path')):
|
|
if package is None:
|
|
checks.append(_doctor_check(name, False, 'not_applicable', 'verified package is unavailable'))
|
|
continue
|
|
try:
|
|
completed = subprocess.run(
|
|
[package[key], '--version'], stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
timeout=10, check=False,
|
|
)
|
|
status = 'ok' if completed.returncode == 0 else 'error'
|
|
checks.append(_doctor_check(name, True, status, f'{name} executable returned {completed.returncode}'))
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
checks.append(_doctor_check(name, True, 'error', f'{name} validation failed: {type(exc).__name__}'))
|
|
if config is None:
|
|
checks.append(_doctor_check('server_reachability', False, 'not_applicable', 'configuration is unavailable'))
|
|
else:
|
|
parsed = urlsplit(config['server'])
|
|
try:
|
|
context = ssl.create_default_context()
|
|
with socket.create_connection((parsed.hostname, parsed.port or 443), timeout=3) as raw:
|
|
with context.wrap_socket(raw, server_hostname=parsed.hostname):
|
|
pass
|
|
checks.append(_doctor_check('server_reachability', True, 'ok', 'server TLS endpoint is reachable'))
|
|
except OSError as exc:
|
|
checks.append(_doctor_check('server_reachability', True, 'error', f'server is unreachable: {type(exc).__name__}'))
|
|
classification = classify_instance(paths['state_dir'])
|
|
status = (
|
|
'ok'
|
|
if classification['state'] in {'running', 'starting', 'draining', 'stopped'}
|
|
else 'error'
|
|
)
|
|
checks.append(_doctor_check('singleton_process', True, status, classification['detail']))
|
|
if _local_layout_ready(paths['state_dir']):
|
|
try:
|
|
local = _open_local_reader(paths)
|
|
retention = local.retention_usage({
|
|
'bundles': paths['bundle_dir'], 'work': paths['work_dir'],
|
|
})
|
|
except (OSError, ValueError, WorkerLocalStateError):
|
|
retention = _fallback_retention(paths)
|
|
checks.append(_doctor_check(
|
|
'local_state', True, 'error', 'local state is unreadable',
|
|
))
|
|
else:
|
|
retention = _fallback_retention(paths)
|
|
checks.append(_doctor_check('retention', True, 'ok', f"{retention['total_bytes']} bytes retained"))
|
|
overall = 'ok' if all(item['status'] in {'ok', 'not_applicable'} for item in checks) else 'error'
|
|
document = {
|
|
'schema': CLI_SCHEMA,
|
|
'command': 'doctor',
|
|
'overall': overall,
|
|
'checks': checks,
|
|
'retention': retention,
|
|
}
|
|
if args.json:
|
|
_print_json(document)
|
|
else:
|
|
print(f'Doctor: {overall}')
|
|
for check in checks:
|
|
print(f"[{check['status']}] {check['name']}: {check['summary']}")
|
|
return EXIT_SUCCESS if overall == 'ok' else EXIT_STARTUP_FAILED
|
|
|
|
|
|
def _load_launch(path, nonce):
|
|
value = read_private_json(path)
|
|
if not isinstance(value, dict) or set(value) != {
|
|
'schema', 'nonce', 'server', 'token', 'parallelism', 'package_manifest',
|
|
'state_dir', 'bundle_dir', 'work_dir', 'retention_days',
|
|
'retention_bytes', 'log_bytes', 'log_files',
|
|
} or value.get('schema') != LAUNCH_SCHEMA or value.get('nonce') != nonce:
|
|
raise InvocationError('detached launch record is invalid')
|
|
_validate_config({
|
|
'schema': CONFIG_SCHEMA,
|
|
'server': value['server'], 'token': value['token'],
|
|
'parallelism': value['parallelism'], 'installed_at': '',
|
|
'retention_days': value['retention_days'],
|
|
'retention_bytes': value['retention_bytes'],
|
|
'log_bytes': value['log_bytes'],
|
|
'log_files': value['log_files'],
|
|
})
|
|
return value
|
|
|
|
|
|
def command_supervise(args):
|
|
try:
|
|
value = _load_launch(args.launch_file, args.launch_nonce)
|
|
durable_unlink(args.launch_file)
|
|
client = SimpleNamespace(
|
|
package_manifest=value['package_manifest'],
|
|
state_dir=value['state_dir'], bundle_dir=value['bundle_dir'],
|
|
work_dir=value['work_dir'], server=value['server'], token=value['token'],
|
|
parallelism=value['parallelism'], poll_seconds=5.0,
|
|
error_delay_seconds=15.0, http_timeout=120,
|
|
retention_days=value['retention_days'],
|
|
retention_bytes=value['retention_bytes'],
|
|
log_bytes=value['log_bytes'], log_files=value['log_files'],
|
|
)
|
|
return WorkerSupervisor(client, foreground=False).run(
|
|
startup_file=args.startup_file, launch_nonce=args.launch_nonce,
|
|
)
|
|
except WorkerAlreadyRunning:
|
|
return EXIT_STARTUP_FAILED
|
|
except BaseException as exc:
|
|
from worker_supervisor import write_startup_result
|
|
try:
|
|
write_startup_result(
|
|
args.startup_file, args.launch_nonce, 'failed', error=type(exc).__name__,
|
|
)
|
|
except OSError:
|
|
pass
|
|
return EXIT_STARTUP_FAILED
|
|
|
|
|
|
def command_assignment_runner(args, paths):
|
|
work_root = canonical_path(ensure_private_directory(
|
|
os.path.abspath(paths['work_dir']), reject_reparse=True,
|
|
))
|
|
root = canonical_path(os.path.abspath(args.root))
|
|
if os.path.dirname(root) != work_root or not os.path.basename(root).startswith(
|
|
'worker-assignment-'
|
|
):
|
|
raise InvocationError('assignment runner root is outside worker work storage')
|
|
package = verify_worker_package(paths['package_manifest'])
|
|
manifest = package.pop('manifest')
|
|
build_compatibility = package.pop('build_compatibility')
|
|
package.pop('runtime_trees')
|
|
runtime = {
|
|
**package,
|
|
'build_compatibility': build_compatibility,
|
|
'capabilities': tuple(
|
|
(item['source'], item['platform'], item['planning_kind'])
|
|
for item in manifest['capabilities']
|
|
),
|
|
}
|
|
return run_assignment(root, runtime)
|
|
|
|
|
|
def main(argv=None):
|
|
try:
|
|
args = parse_args(argv)
|
|
if args.command == '_supervise':
|
|
return command_supervise(args)
|
|
paths = default_worker_paths(module_path=__file__)
|
|
if args.command == '_assignment_runner':
|
|
return command_assignment_runner(args, paths)
|
|
commands = {
|
|
'install': command_install,
|
|
'run': command_run,
|
|
'start': command_start,
|
|
'stop': command_stop,
|
|
'status': command_status,
|
|
'attach': command_attach,
|
|
'watch': command_attach,
|
|
'logs': command_logs,
|
|
'history': command_history,
|
|
'doctor': command_doctor,
|
|
}
|
|
return commands[args.command](args, paths)
|
|
except (InvocationError, ValueError) as exc:
|
|
print(f'truf-worker: {exc}', file=sys.stderr)
|
|
return EXIT_INVALID_INVOCATION
|
|
except (WorkerSupervisorError, WorkerLocalStateError) as exc:
|
|
print(f'truf-worker: {exc}', file=sys.stderr)
|
|
return EXIT_STALE_OR_UNVERIFIABLE
|
|
except KeyboardInterrupt:
|
|
return 130
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|