diff --git a/tests/Makefile.am b/tests/Makefile.am index de8a6eca..eb90d6b6 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -4,13 +4,6 @@ EXTRA_DIST = fixtures/reality_openssl_hello.hex data/silero_speech.pcm data/sile run_tests.py test_run_tests.py test_build_script.py stream_netns/topology.py stream_netns/network.py \ stream_netns/probe.py stream_netns/run.py -if OS_LINUX -# Отдельный привилегированный стенд не запускается обычным make check. -EXTRA_PROGRAMS = stream_netns/stream_lab_node -stream_netns_stream_lab_node_SOURCES = stream_netns/node.c -stream_netns_stream_lab_node_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) -endif - # Порядок по замерам Linux: быстрые проверки раньше длительных; состав зависит от прежних configure-условий. # Быстрые проверки (до 100мс). @@ -192,6 +185,13 @@ noinst_PROGRAMS = EXTRA_PROGRAMS = test_merkle_sync_stress CLEANFILES = test_merkle_sync_stress$(EXEEXT) +if OS_LINUX +# Отдельный привилегированный стенд не запускается обычным make check. +EXTRA_PROGRAMS += stream_netns/stream_lab_node +stream_netns_stream_lab_node_SOURCES = stream_netns/node.c +stream_netns_stream_lab_node_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +endif + # Разрывы двух узлов и race трёх SQLite-баз; не входит в all/check/check.sh. .PHONY: check-merkle-sync-stress check-merkle-sync-stress: test_merkle_sync_stress$(EXEEXT) diff --git a/tests/stream_netns/scenario.py b/tests/stream_netns/scenario.py new file mode 100644 index 00000000..0cc23b16 --- /dev/null +++ b/tests/stream_netns/scenario.py @@ -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