冷启动是一个数字：在你怪罪自己的代码之前，先做一次 90 分钟的主机一致性试点

{ "warm_samples": 30, "cold_samples": 10, "idle_wait_s": 900, "p95_warm_ms": 400, "p95_cold_ms": 1500, "require_digest_parity": true, "max_error_rate": 0.0 } 

#!/usr/bin/env python3 """spike_host_parity.py - one hypothesis, ninety minutes, stdlib only.""" import hashlib import json import sys import time import urllib.request TH = json.load(open("thresholds.json")) JOB = "/run?job=fixture-7" def get(url, timeout=10): t0 = time.perf_counter() with urllib.request.urlopen(url, timeout=timeout) as r: body = r.read() code = r.status return code, body, (time.perf_counter() - t0) * 1000.0 def pct(xs, p): xs = sorted(xs) if not xs: return None k = int(round(p / 100.0 * (len(xs) - 1))) return xs[max(0, min(len(xs) - 1, k))] def digest(body): return hashlib.sha256(body).hexdigest()[:16] def sample(base, n, sleep_s=0.0): lat, codes, digests = [], set(), set() for _ in range(n): try: code, body, ms = get(base + JOB) except Exception as exc: codes.add(type(exc).__name__) continue codes.add(code) lat.append(ms) digests.add(digest(body)) time.sleep(sleep_s) return lat, codes, digests def verdict(): warm = json.load(open("spike_warm.json")) cold = json.load(open("spike_cold.json")) reasons = [] if warm["p95_ms"] is None or cold["p95_ms"] is None: reasons.append("KILL: no samples collected") else: if warm["p95_ms"] > TH["p95_warm_ms"]: reasons.append(f"KILL: warm p95 {warm['p95_ms']} > {TH['p95_warm_ms']}") if cold["p95_ms"] > TH["p95_cold_ms"]: reasons.append(f"KILL: cold p95 {cold['p95_ms']} > {TH['p95_cold_ms']}") if TH["require_digest_parity"] and warm["digests"] != cold["digests"]: reasons.append("KILL: digest mismatch across hosts") if len(warm["digests"]) != 1: reasons.append("KILL: warm digests not stable") print("\n".join(reasons) if reasons else "SHIP: hypothesis survived the window") return 1 if reasons else 0 def main(): if len(sys.argv) < 3: sys.exit("usage: spike_host_parity.py BASE_URL warm|cold|verdict") base, mode = sys.argv[1].rstrip("/"), sys.argv[2] if mode == "verdict": sys.exit(verdict()) n = TH["warm_samples"] if mode == "warm" else TH["cold_samples"] lat, codes, digests = sample(base, n) out = { "phase": mode, "n": len(lat), "codes": sorted(str(c) for c in codes), "digests": sorted(digests), "p50_ms": round(pct(lat, 50), 1) if lat else None, "p95_ms": round(pct(lat, 95), 1) if lat else None, } with open(f"spike_{mode}.json", "w") as fh: json.dump(out, fh, indent=2) print(json.dumps(out, indent=2)) if __name__ == "__main__": main() 

# 1. deploy, then immediately sample the warm path python3 spike_host_parity.py "$HOST_B" warm # 2. wait past the host's idle timeout, then sample the cold path sleep 900 python3 spike_host_parity.py "$HOST_B" cold # 3. verdict reads both files python3 spike_host_parity.py "$HOST_B" verdict