231 lines
12 KiB
Python
231 lines
12 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Monitor one Linux process for a real eight-hour duration/liveness qualification.
|
||
|
|
|
||
|
|
Examples (never selects robot endpoints):
|
||
|
|
python tools/native_soak.py --pid 1234 --output /tmp/soak-unique \
|
||
|
|
--progress-file /tmp/completed.json --journal /tmp/goals.log
|
||
|
|
python tools/native_soak.py --output /tmp/qualification --duration-seconds 30 \
|
||
|
|
--command python -u workload.py
|
||
|
|
|
||
|
|
The workload must atomically replace progress JSON {"completed": N} after actual
|
||
|
|
business work completes. A timer/heartbeat is NOT a valid business progress metric.
|
||
|
|
Counter reset, stale progress, process exit/replacement and sampling failure fail.
|
||
|
|
An attached process is never signaled. A spawned command runs once (no shell or
|
||
|
|
restart); after monitoring it receives SIGTERM, then SIGKILL after five seconds.
|
||
|
|
Only that direct child is owned/monitored: use a foreground executor, not a launcher
|
||
|
|
or shell with detached children. Logs/resources of descendants are not aggregated.
|
||
|
|
|
||
|
|
Exit 0: observed >=8h and progressing; NOT full DR acceptance. Resource growth and
|
||
|
|
workload validity still need review. Exit 2: short/missing-metric/interrupted run.
|
||
|
|
Exit 1: observed failure. Missing final summary means interrupted/incomplete, never
|
||
|
|
pass. No simulated/ROS clock, injectable clock or shortened acceptance threshold.
|
||
|
|
"""
|
||
|
|
import argparse
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
import json
|
||
|
|
import math
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import signal
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
|
||
|
|
EIGHT_HOURS = 8 * 60 * 60
|
||
|
|
|
||
|
|
|
||
|
|
def verdict(elapsed, progress_observed, failure=None, interrupted=False, sampling_gap=False):
|
||
|
|
qualified = (elapsed >= EIGHT_HOURS and progress_observed and not failure
|
||
|
|
and not interrupted and not sampling_gap)
|
||
|
|
return dict(status='failed' if failure else 'duration_liveness_pass' if qualified else 'qualification_incomplete',
|
||
|
|
eight_hour_pass=qualified, no_deadlock_claim=False,
|
||
|
|
duration_liveness_pass=qualified, full_dr_acceptance=False,
|
||
|
|
resource_growth_review_required=True,
|
||
|
|
scope='Same process survival and declared business progress only; not proof of physical safety or absence of all deadlocks')
|
||
|
|
|
||
|
|
|
||
|
|
def positive(text):
|
||
|
|
value = float(text)
|
||
|
|
if not math.isfinite(value) or value <= 0:
|
||
|
|
raise argparse.ArgumentTypeError('must be finite and positive')
|
||
|
|
return value
|
||
|
|
|
||
|
|
|
||
|
|
def process_sample(pid):
|
||
|
|
base = Path('/proc') / str(pid)
|
||
|
|
raw = (base / 'stat').read_text()
|
||
|
|
fields = raw[raw.rfind(')') + 2:].split()
|
||
|
|
if fields[0] in ('Z', 'X', 'x'):
|
||
|
|
raise RuntimeError('process exited (zombie/dead)')
|
||
|
|
status = dict(line.split(':', 1) for line in (base / 'status').read_text().splitlines() if ':' in line)
|
||
|
|
result = dict(pid=pid, start_ticks=int(fields[19]),
|
||
|
|
rss_bytes=int(status['VmRSS'].split()[0]) * 1024,
|
||
|
|
threads=int(status['Threads']), fd_count=len(list((base / 'fd').iterdir())),
|
||
|
|
cpu_seconds=(int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK'))
|
||
|
|
# Detect exit/replacement during multi-file sampling as well as between samples.
|
||
|
|
again = (base / 'stat').read_text().rsplit(')', 1)[1].split()
|
||
|
|
if int(again[19]) != result['start_ticks'] or again[0] in ('Z', 'X', 'x'):
|
||
|
|
raise RuntimeError('process identity changed during sampling')
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def atomic_json(path, value):
|
||
|
|
temporary = path.with_suffix('.tmp')
|
||
|
|
with temporary.open('w') as stream:
|
||
|
|
json.dump(value, stream, indent=2, allow_nan=False)
|
||
|
|
stream.write('\n')
|
||
|
|
stream.flush()
|
||
|
|
os.fsync(stream.fileno())
|
||
|
|
os.replace(temporary, path)
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv=None):
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
|
|
source = parser.add_mutually_exclusive_group(required=True)
|
||
|
|
source.add_argument('--pid', type=int)
|
||
|
|
source.add_argument('--command', nargs=argparse.REMAINDER, help='Foreground argv; must be last option; no shell')
|
||
|
|
parser.add_argument('--output', type=Path, required=True, help='New evidence directory; existing paths refused')
|
||
|
|
parser.add_argument('--duration-seconds', type=positive, default=EIGHT_HOURS)
|
||
|
|
parser.add_argument('--sample-interval-seconds', type=positive, default=5.)
|
||
|
|
parser.add_argument('--max-progress-gap-seconds', type=positive, default=300.)
|
||
|
|
parser.add_argument('--progress-file', type=Path)
|
||
|
|
parser.add_argument('--journal', type=Path, action='append', default=[])
|
||
|
|
args = parser.parse_args(argv)
|
||
|
|
if not sys.platform.startswith('linux'):
|
||
|
|
parser.error('Linux /proc required')
|
||
|
|
if args.pid is not None and args.pid <= 0:
|
||
|
|
parser.error('--pid must be positive')
|
||
|
|
if args.command == []:
|
||
|
|
parser.error('--command requires argv')
|
||
|
|
if args.progress_file and args.sample_interval_seconds >= args.max_progress_gap_seconds:
|
||
|
|
parser.error('sample interval must be shorter than progress gap')
|
||
|
|
args.output.mkdir(parents=True, exist_ok=False)
|
||
|
|
summary_path = args.output / 'summary.json'
|
||
|
|
report = dict(schema_version=1, started_utc=datetime.now(timezone.utc).isoformat(),
|
||
|
|
duration_requested_seconds=args.duration_seconds, acceptance_duration_seconds=EIGHT_HOURS,
|
||
|
|
clock='time.monotonic (host Linux; excludes suspend)',
|
||
|
|
mode='attach' if args.pid else 'spawn', command=args.command,
|
||
|
|
progress_file=str(args.progress_file) if args.progress_file else None,
|
||
|
|
progress_contract='Atomic JSON completed counter incremented by actual workload completion',
|
||
|
|
sample_interval_seconds=args.sample_interval_seconds,
|
||
|
|
max_progress_gap_seconds=args.max_progress_gap_seconds,
|
||
|
|
boot_id=Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
|
||
|
|
status='running', eight_hour_pass=False, full_dr_acceptance=False,
|
||
|
|
resource_growth_review_required=True)
|
||
|
|
atomic_json(summary_path, report)
|
||
|
|
child = None
|
||
|
|
interrupted = False
|
||
|
|
failure = None
|
||
|
|
gap = False
|
||
|
|
start = time.monotonic()
|
||
|
|
last_progress = start
|
||
|
|
previous_counter = None
|
||
|
|
progress_observed = False
|
||
|
|
previous = None
|
||
|
|
identity = None
|
||
|
|
sample_count = 0
|
||
|
|
peaks = dict(rss_bytes=0, threads=0, fd_count=0)
|
||
|
|
old_handlers = {}
|
||
|
|
|
||
|
|
def request_stop(signum, frame):
|
||
|
|
nonlocal interrupted
|
||
|
|
interrupted = True
|
||
|
|
|
||
|
|
try:
|
||
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
||
|
|
old_handlers[sig] = signal.signal(sig, request_stop)
|
||
|
|
with (args.output / 'process.log').open('w') as process_log, (args.output / 'samples.jsonl').open('w') as samples:
|
||
|
|
if args.command:
|
||
|
|
child = subprocess.Popen(args.command, stdin=subprocess.DEVNULL, stdout=process_log, stderr=subprocess.STDOUT)
|
||
|
|
pid = child.pid if child else args.pid
|
||
|
|
report['pid'] = pid
|
||
|
|
start = time.monotonic()
|
||
|
|
last_progress = start
|
||
|
|
while not interrupted:
|
||
|
|
if child and child.poll() is not None:
|
||
|
|
raise RuntimeError('process exited before observation completed: ' + str(child.returncode))
|
||
|
|
sample = process_sample(pid)
|
||
|
|
now = time.monotonic()
|
||
|
|
sample['elapsed_seconds'] = now - start
|
||
|
|
if identity is None:
|
||
|
|
identity = sample['start_ticks']
|
||
|
|
report['start_ticks'] = identity
|
||
|
|
if sample['start_ticks'] != identity:
|
||
|
|
raise RuntimeError('process identity changed (PID reused)')
|
||
|
|
sample['cpu_percent'] = None
|
||
|
|
if previous:
|
||
|
|
delta = sample['elapsed_seconds'] - previous['elapsed_seconds']
|
||
|
|
sample['cpu_percent'] = 100 * (sample['cpu_seconds'] - previous['cpu_seconds']) / delta
|
||
|
|
if delta > max(5., args.sample_interval_seconds * 3):
|
||
|
|
gap = True
|
||
|
|
sample['journal_bytes'] = {str(p.resolve()): p.stat().st_size for p in args.journal}
|
||
|
|
sample['completed'] = None
|
||
|
|
if args.progress_file:
|
||
|
|
# Startup and subsequent progress share the same strict bound.
|
||
|
|
# Late evidence cannot retroactively erase an observed gap.
|
||
|
|
if now - last_progress > args.max_progress_gap_seconds:
|
||
|
|
raise RuntimeError('progress deadline exceeded; process liveness is insufficient')
|
||
|
|
try:
|
||
|
|
value = json.loads(args.progress_file.read_text())['completed']
|
||
|
|
except FileNotFoundError:
|
||
|
|
if previous_counter is not None:
|
||
|
|
raise RuntimeError('progress metric disappeared')
|
||
|
|
value = None # Allow initial workload startup, bounded by the same deadline.
|
||
|
|
if value is not None:
|
||
|
|
if type(value) is not int or value < 0:
|
||
|
|
raise RuntimeError('progress counter must be a nonnegative integer')
|
||
|
|
if previous_counter is not None:
|
||
|
|
if value < previous_counter:
|
||
|
|
raise RuntimeError('progress counter regressed')
|
||
|
|
if value > previous_counter:
|
||
|
|
progress_observed = True
|
||
|
|
last_progress = now
|
||
|
|
previous_counter = value
|
||
|
|
sample['completed'] = value
|
||
|
|
samples.write(json.dumps(sample, allow_nan=False) + '\n')
|
||
|
|
samples.flush()
|
||
|
|
os.fsync(samples.fileno())
|
||
|
|
sample_count += 1
|
||
|
|
for name in peaks:
|
||
|
|
peaks[name] = max(peaks[name], sample[name])
|
||
|
|
previous = sample
|
||
|
|
report.update(elapsed_seconds=now - start, samples=sample_count, resource_peaks=peaks,
|
||
|
|
progress_observed=progress_observed, last_completed=previous_counter)
|
||
|
|
atomic_json(summary_path, report)
|
||
|
|
if now - start >= args.duration_seconds:
|
||
|
|
break
|
||
|
|
deadline = min(now + args.sample_interval_seconds, start + args.duration_seconds)
|
||
|
|
while not interrupted and time.monotonic() < deadline:
|
||
|
|
time.sleep(max(0., min(1., deadline - time.monotonic())))
|
||
|
|
except (Exception, KeyboardInterrupt) as exc:
|
||
|
|
failure = str(exc) or type(exc).__name__
|
||
|
|
finally:
|
||
|
|
elapsed = previous['elapsed_seconds'] if previous else 0.
|
||
|
|
report.update(verdict(elapsed, progress_observed, failure, interrupted, gap))
|
||
|
|
report.update(elapsed_seconds=elapsed, reason=failure or ('interrupted' if interrupted else 'monitoring completed'),
|
||
|
|
sampling_gap=gap, samples=sample_count, progress_observed=progress_observed,
|
||
|
|
resource_peaks=peaks, ended_utc=datetime.now(timezone.utc).isoformat(),
|
||
|
|
child_cleanup='not applicable; attached process untouched')
|
||
|
|
if child:
|
||
|
|
if child.poll() is None:
|
||
|
|
child.terminate()
|
||
|
|
try:
|
||
|
|
child.wait(timeout=5)
|
||
|
|
report['child_cleanup'] = 'direct child terminated after monitoring'
|
||
|
|
except subprocess.TimeoutExpired:
|
||
|
|
child.kill()
|
||
|
|
child.wait()
|
||
|
|
report['child_cleanup'] = 'direct child killed after five-second termination timeout'
|
||
|
|
else:
|
||
|
|
report['child_cleanup'] = 'direct child had already exited'
|
||
|
|
report['child_exit_code'] = child.returncode
|
||
|
|
for sig, old in old_handlers.items():
|
||
|
|
signal.signal(sig, old)
|
||
|
|
atomic_json(summary_path, report)
|
||
|
|
print(json.dumps(report))
|
||
|
|
return 1 if failure else 0 if report['duration_liveness_pass'] else 2
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
raise SystemExit(main())
|