"""Управление настоящими процессами uTun; сетевые пакеты не подменяются.""" import csv import json from pathlib import Path import select import subprocess import time from topology import NODES, WANS class Agent: def __init__(self, network, node, binary, seed, tcp, tcp_only): self.node = node directory = network.directory / node.name directory.mkdir() (directory / "db").mkdir() identity = json.loads(subprocess.check_output([str(binary), "--identity", f"stream-lab:{seed}:{node.name}"], text=True)) self.identity = identity["node"] kind = "nat" if node.nat else "public" sockets = "".join(f"[server:{transport}]\naddr={node.ip}:32000\ntype={kind}\ntransport={transport}\n" for transport in (["tcp"] if tcp_only else ["udp", "tcp"] if tcp else ["udp"])) config = (f"[global]\nmy_node_name={node.name}\nmy_private_key={identity['private']}\n" f"my_public_key={identity['public']}\nclient_type={node.client_type}\nauto_sockets=no\n" f"db_path={directory / 'db'}\ndebug_level=warn\n" "[ntp]\nenabled=no\n[allowed_keys]\nallow_all=1\n" "[chatserver]\ngroup_autoconnect=1\nstorage_autoload=0\nmember_files_enabled=0\n" "[debug]\nstream=debug\nchat_sync=info\nmember_sync=info\nbgp=info\nconnection=info\n" + sockets) path = directory / "node.conf" path.write_text(config) self.stderr = (directory / "stderr.log").open("w") self.transcript = (directory / "rpc.jsonl").open("w", buffering=1) self.process = network.spawn(node.name, [str(binary), str(path), str(directory / "utun.log")], stdin=-1, stdout=-1, stderr=self.stderr, text=True, bufsize=1) self.serial = 0 if self.read(10) != {"ready": True, "node": self.identity}: raise RuntimeError(f"{node.name}: unexpected identity/ready response") def read(self, timeout): if not select.select([self.process.stdout], [], [], timeout)[0]: raise RuntimeError(f"{self.node.name}: RPC timed out ({timeout}s); see utun.log") line = self.process.stdout.readline() self.transcript.write(json.dumps({"direction": "out", "time": time.monotonic(), "line": line.rstrip()}) + "\n") if not line: raise RuntimeError(f"{self.node.name}: agent exited rc={self.process.poll()}; see stderr.log/utun.log") try: return json.loads(line) except json.JSONDecodeError as error: raise RuntimeError(f"{self.node.name}: non-JSON stdout: {line[:200]}") from error def rpc(self, op, *, timeout=5, allow_wait=False, **arguments): self.serial += 1 request = {key: str(value) for key, value in dict(arguments, op=op, id=self.serial).items()} self.transcript.write(json.dumps({"direction": "in", "time": time.monotonic(), "request": request}) + "\n") self.process.stdin.write(json.dumps(request) + "\n") self.process.stdin.flush() response = self.read(timeout) if response.get("id") != self.serial or response.get("rc") not in ({0, 2} if allow_wait else {0}): raise RuntimeError(f"{self.node.name}: {op} failed: {response}") return response def close(self): if self.process.poll() is None: try: self.rpc("quit") self.process.wait(timeout=10) except Exception: self.process.kill() self.process.wait() self.stderr.close() self.transcript.close() return self.process.returncode class Scenario: def __init__(self, network, args): self.network, self.args = network, args self.agents = {} self.samples = (network.directory / "samples.jsonl").open("w", buffering=1) self.started = time.monotonic() self.phases = [] self.latest = {} self.last_metadata = {} self.clock_previous = {} self.clock_wraps = 0 def sample(self, phase): snapshot = {name: agent.rpc("stats") for name, agent in self.agents.items()} self.latest = snapshot self.samples.write(json.dumps({"elapsed_s": time.monotonic() - self.started, "phase": phase, "nodes": snapshot}) + "\n") for name, statistics in snapshot.items(): if statistics["errors"] or any(s["errors"] for s in statistics["streams"]): raise RuntimeError(f"{name}: payload/order validation failed; see samples.jsonl and utun.log") if any(s.get("stored_bytes", 0) > 32 * 1024 * 1024 for s in statistics["streams"]): raise RuntimeError(f"{name}: stream data buffer exceeded 32 MiB") for stream in statistics["streams"]: self.last_metadata.setdefault((name, stream["source"]), {}).update(stream) path = next((p for p in stream.get("paths", []) if p["sub"] == 0), {}) if stream["publisher"] and "source_timestamp_ms" in path: key = (name, stream["source"]) current = (path["seq"], path["source_timestamp_ms"]) previous = self.clock_previous.get(key) if previous and current[0] > previous[0] and current[1] < previous[1]: self.clock_wraps += 1 self.clock_previous[key] = current return snapshot def wait(self, phase, predicate, timeout): deadline = time.monotonic() + timeout while time.monotonic() < deadline: statistics = self.sample(phase) if predicate(statistics): return statistics time.sleep(1) raise RuntimeError(f"{phase}: condition not met in {timeout}s; see last samples.jsonl record") def duration(self, phase, seconds): print(f"Phase {phase}: {seconds}s", flush=True) started = time.monotonic() self.phases.append({"phase": phase, "begin_s": started - self.started, "duration_s": seconds}) while time.monotonic() - started < seconds: self.sample(phase) time.sleep(1) @staticmethod def stream(statistics, source): return next((s for s in statistics["streams"] if s["source"] == source), {}) def ready(self, statistics, source, names): return all(self.stream(statistics[name], source).get("received", 0) >= 10 and self.stream(statistics[name], source).get("age_ms", 100000) < 5000 for name in names) def run(self): binary = Path(self.args.binary).resolve() for node in NODES: self.agents[node.name] = Agent(self.network, node, binary, self.args.seed, self.args.tcp, self.args.tcp_only) owner = self.agents["n01"] group = owner.rpc("create")["group"] print(f"Joining real CHAT group {group}", flush=True) ordered = [n for n in NODES if not n.nat and n.name != "n01"] + [n for n in NODES if n.nat] for node in ordered: link = owner.rpc("invite", timeout=15)["link"] self.agents[node.name].rpc("join", link=link) identities = {agent.identity for agent in self.agents.values()} self.wait("join-merkle-bgp", lambda s: all(set(n["online"]) == identities and set(n["profile_status"]) == identities and all(rc == 0 for rc in n["profile_status"].values()) for n in s.values()), 120) source = self.agents[self.args.publisher] source_id = source.identity source.rpc("publish", seed=self.args.seed) pending = {name for name in self.agents if name != self.args.publisher} deadline = time.monotonic() + 90 while pending and time.monotonic() < deadline: for name in list(pending): if self.agents[name].rpc("subscribe", source=source_id, allow_wait=True)["rc"] == 0: pending.remove(name) if pending: time.sleep(1) if pending: raise RuntimeError(f"Signed source profile did not reach subscribers: {sorted(pending)}") source.rpc("rate", source=source_id, bps=self.args.rate) names = set(self.agents) - {self.args.publisher} self.wait("initial-delivery", lambda s: self.ready(s, source_id, names), 90) self.duration("baseline", self.args.duration) if self.args.faults: link = WANS[1] self.network.bandwidth(link, 2) for node in NODES: if node.country in {link.a, link.b}: other = link.b if node.country == link.a else link.a self.network.netem(node, other, rtt=300, loss=5) self.duration("wan-degraded", self.args.fault_duration) before_recovery = {name: self.stream(self.latest[name], source_id)["received"] for name in names} self.network.bandwidth(link, link.mbps) for node in NODES: if node.country in {link.a, link.b}: self.network.netem(node, link.b if node.country == link.a else link.a) self.wait("wan-recovery", lambda s: self.ready(s, source_id, names) and all(self.stream(s[name], source_id)["received"] > before_recovery[name] for name in names), 90) self.duration("recovered", 10) # Узлы без подписки должны прекратить обработку этого экземпляра. excluded = [name for name in ("n02", "n13") if name != self.args.publisher] for name in excluded: self.agents[name].rpc("close", source=source_id) counts = {name: self.stream(self.agents[name].rpc("stats"), source_id)["received"] for name in excluded} self.duration("unsubscribed", 5) for name in excluded: if self.stream(self.latest[name], source_id)["received"] != counts[name]: raise RuntimeError(f"{name}: received DATA after subscription close") names -= set(excluded) source.rpc("finish", source=source_id) final = self.wait("signed-end", lambda s: all(self.stream(s[n], source_id).get("ends", 0) for n in names), 60) generated = self.stream(final[self.args.publisher], source_id)["generated"] if generated < 10: raise RuntimeError("Publisher generated insufficient data") if self.args.duration >= 66 and not self.clock_wraps: raise RuntimeError("Full-duration run did not observe a source timestamp wrap") for name in names: stream = self.stream(final[name], source_id) if stream["next_seq"] != generated: raise RuntimeError(f"{name}: END delivered before final cursor: {stream['next_seq']} != {generated}") if stream["received"] + stream["gap_batches"] + stream["start_seq"] != generated: raise RuntimeError(f"{name}: unaccounted records: received={stream['received']}, gaps={stream['gap_batches']}, " f"start={stream['start_seq']}, generated={generated}") self.report(final, source_id) return {"pass": True, "group": group, "publisher": self.args.publisher, "generated": generated, "phases": self.phases, "subscribers_checked": len(names), "source_timestamp_wraps": self.clock_wraps, "transport": "tcp" if self.args.tcp_only else "udp+tcp" if self.args.tcp else "udp", "rate_bps": self.args.rate} def report(self, final, source): rows = [] for node in NODES: stream = self.stream(final[node.name], source) metadata = self.last_metadata[(node.name, source)] rows.append(dict(node=node.name, country=node.country, nat=node.nat or "public", received=stream["received"], gap_batches=stream["gap_batches"], errors=stream["errors"], p50_ms=stream["latency_p50_ms"], p95_ms=stream["latency_p95_ms"], p99_ms=stream["latency_p99_ms"], maximum_ms=stream["latency_max_ms"], migrations=metadata.get("migrations", 0), probes=metadata.get("probes", 0))) with (self.network.directory / "summary.csv").open("w", newline="") as output: writer = csv.DictWriter(output, fieldnames=list(rows[0])) writer.writeheader(); writer.writerows(rows) (self.network.directory / "final-stats.json").write_text(json.dumps(final, indent=2) + "\n") def close(self): (self.network.directory / "last-stats.json").write_text(json.dumps(self.latest, indent=2) + "\n") failures = {} for name, agent in self.agents.items(): status = agent.close() if status: failures[name] = status self.samples.close() return failures