evgeny
2 days ago
2 changed files with 218 additions and 7 deletions
@ -0,0 +1,211 @@
|
||||
"""Управление настоящими процессами 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): |
||||
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 (["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 = {} |
||||
|
||||
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") |
||||
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) |
||||
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 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) |
||||
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), 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") |
||||
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["first_seq"] != generated: |
||||
raise RuntimeError(f"{name}: unaccounted missing records") |
||||
self.report(final, source_id) |
||||
return {"pass": True, "group": group, "publisher": self.args.publisher, "generated": generated, |
||||
"phases": self.phases, "subscribers_checked": len(names)} |
||||
|
||||
def report(self, final, source): |
||||
rows = [] |
||||
for node in NODES: |
||||
stream = self.stream(final[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=stream.get("migrations", 0), probes=stream.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): |
||||
failures = {} |
||||
for name, agent in self.agents.items(): |
||||
status = agent.close() |
||||
if status: |
||||
failures[name] = status |
||||
self.samples.close() |
||||
return failures |
||||
Loading…
Reference in new issue