App 加固后内存峰值与 OOM 风险怎样验收

import json
import statistics
import sys
from collections import defaultdict
from pathlib import Path

if len(sys.argv) != 2:
    raise SystemExit('usage: aggregate_memory_samples.py samples.jsonl')
input_path = Path(sys.argv[1])
if not input_path.is_file():
    raise SystemExit('sample file is missing')
groups = defaultdict(list)
for line_number, line in enumerate(input_path.read_text(encoding='utf-8').splitlines(), 1):
    if not line.strip():
        continue
    try:
        row = json.loads(line)
    except json.JSONDecodeError as exc:
        raise SystemExit(f'invalid JSON at line {line_number}: {exc}')
    for field in ('run_id', 'scenario', 'timestamp_ms', 'pss_kb'):
        if field not in row:
            raise SystemExit(f'missing {field} at line {line_number}')
    if not isinstance(row['timestamp_ms'], int) or row['timestamp_ms'] < 0:
        raise SystemExit(f'invalid timestamp at line {line_number}')
    if not isinstance(row['pss_kb'], (int, float)) or row['pss_kb'] < 0:
        raise SystemExit(f'invalid PSS at line {line_number}')
    groups[(str(row['run_id']), str(row['scenario']))].append(row)
if not groups:
    raise SystemExit('no samples found')
results = []
for (run_id, scenario), rows in sorted(groups.items()):
    rows.sort(key=lambda item: item['timestamp_ms'])
    if len(rows) < 3:
        raise SystemExit(f'too few samples for {run_id}/{scenario}')
    timestamps = [item['timestamp_ms'] for item in rows]
    if len(set(timestamps)) != len(timestamps):
        raise SystemExit(f'duplicate timestamps for {run_id}/{scenario}')
    values = [float(item['pss_kb']) for item in rows]
    tail = values[-max(3, len(values) // 3):]
    exits = sorted({str(item['exit_reason']) for item in rows if item.get('exit_reason')})
    results.append({'run_id': run_id, 'scenario': scenario, 'samples': len(rows), 'peak_pss_kb': max(values), 'median_pss_kb': statistics.median(values), 'tail_median_pss_kb': statistics.median(tail), 'exit_reasons': exits, 'gaps': [] if exits else ['exit-reason-not-observed']})
print(json.dumps(results, ensure_ascii=False, indent=2))