mirror of
https://github.com/jmcorgan/fips.git
synced 2026-07-30 19:46:15 +00:00
testing: add bloom-storm chaos scenario
Six-node depth-4 mesh with an induced upstream parent flap. Asserts a
trailing-window ceiling on per-node `stats.bloom.sent` and a sanity
floor on parent-switch count over a ~3-4 min observation window.
Guards against the regression class where a spanning-tree update that
changes only an internal path edge (no root or depth delta) fails to
be properly contained and instead propagates to leaves as a sustained
bloom-traffic oscillation, visible only at fleet scale and only after
several minutes of uptime.
Adds a new chaos primitive (`link_swap`) for deterministic asymmetric
link-cost flapping and a post-run assertion framework with two
checks:
- `bloom_send_rate.max_per_node`: trailing-window ceiling on the
`show_bloom` stats counter delta. Calibrated against the
post-mortem reproduction harness data (per-variant counter table
against pre-fix vs post-fix binaries).
- `min_parent_switches.min_total`: sanity guard against a
misconfigured harness where the flap inducer fires but the
topology never produces a real parent-switch event (e.g., wrong
root election from a different seed). Without this, the
bloom-rate assertion would trivially pass on any binary
including a regressed one.
The runner exits 3 on assertion failure (alongside 0 success and 2
panic-detected). Threshold derivation is documented in the scenario
README; the seed pin is also documented there since smallest-NodeAddr
root election is sensitive to the pubkey hash ordering.
Wired into ci-local.sh's chaos pool and the GitHub CI chaos matrix.
This commit is contained in:
@@ -366,6 +366,9 @@ jobs:
|
||||
- suite: congestion-stress
|
||||
type: chaos
|
||||
scenario: congestion-stress
|
||||
- suite: bloom-storm
|
||||
type: chaos
|
||||
scenario: bloom-storm
|
||||
# ── Sidecar deployment ──────────────────────────────────────────
|
||||
- suite: sidecar
|
||||
type: sidecar
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
# bloom-storm — bloom announce storm regression scenario
|
||||
|
||||
Six-node depth-4 mesh with an induced upstream parent flap. Asserts a
|
||||
trailing-window ceiling on per-node `stats.bloom.sent` to catch a
|
||||
regression class where a localized spanning-tree update at an
|
||||
internal mid-chain edge fails to be properly contained and instead
|
||||
propagates as a bloom announce storm down to the leaves.
|
||||
|
||||
## Topology
|
||||
|
||||
```text
|
||||
n01 (root, depth 0)
|
||||
/ \
|
||||
n02 n03 (parent candidates pa/pb, depth 1)
|
||||
\ /
|
||||
n04 (flap, depth 2 — induced parent flap)
|
||||
|
|
||||
n05 (leaf, depth 3)
|
||||
|
|
||||
n06 (tail, depth 4)
|
||||
```
|
||||
|
||||
n01 is expected to win the root election under the deterministic key
|
||||
derivation used by the chaos runner; the run's tree snapshots
|
||||
(`tree-snapshot-warmup.json`, `tree-snapshot-final.json`) record the
|
||||
actual assignment.
|
||||
|
||||
## Bug class guarded against
|
||||
|
||||
A spanning-tree update that changes only an internal path edge — no
|
||||
root change, no depth change — must not produce a sustained bloom
|
||||
announce storm at downstream nodes. The original regression
|
||||
(rolled-back `0caef2a`, fixed in master `4cdf382`) had this property:
|
||||
in the field, a single mid-chain ancestor swap on an upstream node
|
||||
caused every downstream node in its subtree to issue a bloom
|
||||
announce on every parent re-evaluation tick of the upstream node,
|
||||
resulting in a ~480x mesh-wide elevation in `FilterAnnounce` traffic
|
||||
and a perfect bimodal flip on `est_entries`.
|
||||
|
||||
## Mechanism
|
||||
|
||||
The new `link_swap` chaos primitive deterministically alternates the
|
||||
netem delay on `n02-n04` (5ms vs 100ms) and `n03-n04` (100ms vs 5ms)
|
||||
every 4 seconds. Combined with the FIPS overrides
|
||||
(`parent_hysteresis: 0.0`, `reeval_interval_secs: 1`,
|
||||
`flap_threshold: 9999`, `hold_down_secs: 0`), this forces n04 to
|
||||
switch parents on every swap. The whole point of the scenario is to
|
||||
*force* sustained parent flapping at n04 and assert that the bloom
|
||||
layer doesn't amplify it.
|
||||
|
||||
## Assertions
|
||||
|
||||
```yaml
|
||||
assertions:
|
||||
bloom_send_rate:
|
||||
window_secs: 30
|
||||
max_per_node: 30
|
||||
min_parent_switches:
|
||||
min_total: 10
|
||||
```
|
||||
|
||||
`bloom_send_rate` is the load-bearing assertion: per-node delta of
|
||||
`stats.bloom.sent` over the trailing 30s of the run must be at most
|
||||
30. Per-node deltas and the offending node IDs are written to
|
||||
`assertions.txt` and the runner exits 3 on failure.
|
||||
|
||||
`min_parent_switches` is a sanity guard. It fails if the run did not
|
||||
record at least 10 parent switches across all nodes, which would
|
||||
mean the harness fired its link swaps but the topology never produced
|
||||
a real parent-switch event (e.g., wrong root election made the flap
|
||||
target's parent candidates structurally non-equivalent). Without this
|
||||
guard, the bloom-rate assertion would trivially pass on any binary,
|
||||
including a regressed one.
|
||||
|
||||
## Threshold derivation
|
||||
|
||||
The original `issues/2026-0019-repro/` reproduction harness measured
|
||||
(90s flap window, ~21 induced parent switches at the mid-chain
|
||||
node):
|
||||
|
||||
| binary | tail bloom_sent / 90s | rate scaled to 30s |
|
||||
| ----------- | --------------------: | -----------------: |
|
||||
| pre-fix | 21 | ~7 |
|
||||
| current fix | 0 | 0 |
|
||||
|
||||
Observed on this scenario at master `db5b6b1` (180s run, 35 parent
|
||||
switches, 41 link swaps), per-node `bloom_sent` deltas over the
|
||||
trailing 30s:
|
||||
|
||||
```text
|
||||
n01=5 n02=5 n03=4 n04=12 n05=6 n06=0
|
||||
```
|
||||
|
||||
n04 (the flapping node) is the highest because it is legitimately
|
||||
re-sending its filter on its own parent changes. n06 (the depth-4
|
||||
"tail") sees 0, matching the calm post-fix behavior recorded in
|
||||
`issues/2026-0019-repro/RESULTS.md` for the `fix2` variant.
|
||||
|
||||
In the field, the regression's mesh-wide rate scaled ~480x above
|
||||
steady state. A `30 / 30s / node` ceiling sits ~2.5x above the
|
||||
observed maximum on fixed master and well below the
|
||||
deployment-scale storm rate, giving headroom for harness jitter
|
||||
without losing the ability to fail loud on the regression class.
|
||||
|
||||
If `link_swap.interval_secs` or the netem delta is changed,
|
||||
recalibrate. The threshold is calibrated against the values in
|
||||
`scenarios/bloom-storm.yaml` as committed and the `seed: 31` pin.
|
||||
|
||||
## Limitations
|
||||
|
||||
- The bloom-storm regression has not been confirmed-failing here
|
||||
on a regressed binary in this harness directly; the threshold is
|
||||
inferred from the values measured in the dedicated
|
||||
`issues/2026-0019-repro/` post-mortem harness against
|
||||
`0caef2a`. To gain that confirmation, check out `0caef2a`
|
||||
(or the `backup-broadcast-gate-bloom-storm` branch if still
|
||||
retained), build, copy binaries into `testing/docker/`, and rerun
|
||||
this scenario; the bloom-rate assertion is expected to fail loud
|
||||
with n05/n06 deltas well above 30.
|
||||
|
||||
- Root-election outcome is sensitive to the seed (smallest
|
||||
`NodeAddr` wins, where `NodeAddr = SHA-256(pubkey)[..16]`). The
|
||||
seed value `31` is pinned for this reason. The
|
||||
`min_parent_switches` assertion catches drift if the seed is
|
||||
changed without re-validating the topology.
|
||||
|
||||
## Running locally
|
||||
|
||||
```bash
|
||||
# From the source repo root, with binaries already built and copied
|
||||
# into testing/docker/ (see testing/scripts/build.sh).
|
||||
./testing/chaos/scripts/chaos.sh bloom-storm
|
||||
```
|
||||
|
||||
Run output is in `testing/chaos/sim-results/<timestamp>-bloom-storm/`.
|
||||
Key artifacts:
|
||||
|
||||
- `analysis.txt` — log analysis (panics, errors, parent switches).
|
||||
- `assertions.txt` — per-assertion pass/fail with per-node deltas.
|
||||
- `tree-snapshot-warmup.json`, `tree-snapshot-final.json` — control
|
||||
socket tree state at warmup end and at run end.
|
||||
- `runner.log` — full orchestration log.
|
||||
|
||||
Total runtime: ~3.5 minutes (25s warmup + 180s scenario + ~30s
|
||||
teardown).
|
||||
@@ -0,0 +1,142 @@
|
||||
# Bloom-storm regression scenario.
|
||||
#
|
||||
# Six-node depth-4 mesh with an induced upstream parent flap. Guards
|
||||
# against a regression class where a spanning-tree update that changes
|
||||
# only an internal path edge (no root or depth delta) fails to
|
||||
# propagate to leaves, producing a sustained bloom-traffic
|
||||
# oscillation visible only at fleet scale and only after several
|
||||
# minutes of uptime. See README.md alongside this file for the
|
||||
# bug-class description and threshold derivation.
|
||||
#
|
||||
# Topology (n01 should win the root election under deterministic
|
||||
# key derivation; the validation snapshot at end-of-run will confirm
|
||||
# the assigned root):
|
||||
#
|
||||
# n01 (root, depth 0)
|
||||
# / \
|
||||
# n02 n03 (parent candidates pa/pb, depth 1)
|
||||
# \ /
|
||||
# n04 (flap, depth 2 — induced parent flap)
|
||||
# |
|
||||
# n05 (leaf, depth 3 — observes mid-chain swap)
|
||||
# |
|
||||
# n06 (tail, depth 4 — sees the cascade)
|
||||
#
|
||||
# The flap is induced by alternating the netem delay on n02-n04 and
|
||||
# n03-n04 every 4 seconds. Because the FIPS overrides below disable
|
||||
# parent-flap dampening, n04 switches parents on each swap. The bug
|
||||
# class under test is whether this localized swap at depth 2 leaks
|
||||
# downstream as a bloom-announce storm at depth 3 and depth 4.
|
||||
#
|
||||
# Assertions (bloom_send_rate.max_per_node_per_window) are evaluated
|
||||
# over the trailing 30s of the run, after sustained flapping has had
|
||||
# time to expose any cascade behavior.
|
||||
|
||||
scenario:
|
||||
name: "bloom-storm"
|
||||
# Seed pinned so deterministic key derivation makes n01 win the
|
||||
# root election and orders the other NodeAddrs as n02 < n03 <
|
||||
# ... < n04 (so n04 is at the bottom of the tiebreak and its
|
||||
# parent candidates n02/n03 are both at depth 1). The warmup tree
|
||||
# snapshot in sim-results/ is the authoritative check.
|
||||
#
|
||||
# Other seeds may elect a different node as root or order n04
|
||||
# ahead of n02/n03, in which case the induced flap will not
|
||||
# exercise the cascade path the assertion is written to catch.
|
||||
seed: 31
|
||||
duration_secs: 180
|
||||
|
||||
topology:
|
||||
algorithm: explicit
|
||||
num_nodes: 6
|
||||
params:
|
||||
adjacency:
|
||||
- [n01, n02]
|
||||
- [n01, n03]
|
||||
- [n02, n04]
|
||||
- [n03, n04]
|
||||
- [n04, n05]
|
||||
- [n05, n06]
|
||||
subnet: "172.20.0.0/24"
|
||||
ip_start: 10
|
||||
|
||||
netem:
|
||||
enabled: true
|
||||
default_policy:
|
||||
# Calm baseline so steady-state bloom traffic is at its floor.
|
||||
delay_ms: [1, 1]
|
||||
jitter_ms: [0, 0]
|
||||
loss_pct: [0, 0]
|
||||
|
||||
# Asymmetric link-cost flap on n04's two upstream candidates. Every
|
||||
# `interval_secs`, the policies on the two listed edges are swapped.
|
||||
# This drives n04 to alternate parents between n02 and n03 each
|
||||
# round. The 4s cadence and 5ms-vs-100ms delta come from the
|
||||
# original ISSUE-2026-0019 reproduction harness — they are
|
||||
# calibrated to produce a parent switch per round under the
|
||||
# zero-hysteresis FIPS overrides below.
|
||||
link_swap:
|
||||
enabled: true
|
||||
interval_secs: 4
|
||||
policies:
|
||||
fast:
|
||||
delay_ms: [5, 5]
|
||||
jitter_ms: [0, 0]
|
||||
loss_pct: [0, 0]
|
||||
slow:
|
||||
delay_ms: [100, 100]
|
||||
jitter_ms: [0, 0]
|
||||
loss_pct: [0, 0]
|
||||
# Initial assignment: n02-n04 fast, n03-n04 slow. After each
|
||||
# interval the assignments are swapped.
|
||||
edges:
|
||||
- edge: "n02-n04"
|
||||
policy: fast
|
||||
- edge: "n03-n04"
|
||||
policy: slow
|
||||
|
||||
link_flaps:
|
||||
enabled: false
|
||||
|
||||
traffic:
|
||||
enabled: false
|
||||
|
||||
logging:
|
||||
rust_log: "info"
|
||||
output_dir: "./sim-results"
|
||||
|
||||
# Per-node assertions evaluated at the end of the run. Each asserts
|
||||
# on the delta of a control-socket counter over a trailing window.
|
||||
assertions:
|
||||
bloom_send_rate:
|
||||
# The trailing window is the last `window_secs` of the run.
|
||||
# `max_per_node` is the ceiling on stats.bloom.sent delta over
|
||||
# that window, per node. The threshold is set ~5x above the
|
||||
# observed steady-state ceiling on fixed master and well below
|
||||
# the storm-state value on the regressed binary.
|
||||
window_secs: 30
|
||||
max_per_node: 30
|
||||
min_parent_switches:
|
||||
# Sanity guard: detect a misconfigured harness where the flap
|
||||
# inducer fires but the topology never produces a real parent
|
||||
# switch (e.g., wrong root election made n04's parent
|
||||
# candidates structurally non-equivalent). Without this, the
|
||||
# bloom-rate assertion would trivially pass on any binary,
|
||||
# including a regressed one.
|
||||
#
|
||||
# Floor calibrated against expected behavior under the chosen
|
||||
# seed: 41 link swaps over 180s should produce >= 10 parent
|
||||
# switches at n04 alone if the topology is right.
|
||||
min_total: 10
|
||||
|
||||
# FIPS overrides to suppress parent-flap dampening so the asymmetric
|
||||
# link-cost flap reliably produces a parent switch per round. The
|
||||
# whole point of the scenario is to *force* sustained parent
|
||||
# flapping at n04 and assert that bloom layer doesn't amplify it.
|
||||
fips_overrides:
|
||||
node:
|
||||
tree:
|
||||
parent_hysteresis: 0.0
|
||||
reeval_interval_secs: 1
|
||||
flap_threshold: 9999
|
||||
hold_down_secs: 0
|
||||
@@ -54,6 +54,8 @@ def main():
|
||||
|
||||
if result and result.panics:
|
||||
sys.exit(2)
|
||||
if runner.assertions_failed:
|
||||
sys.exit(3)
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Post-run scenario assertions evaluated via control-socket data.
|
||||
|
||||
Assertions are declared in the scenario YAML under ``assertions:`` and
|
||||
are evaluated near the end of the simulation, before teardown begins.
|
||||
Each failing assertion is recorded with a clear pass/fail message; the
|
||||
runner exits non-zero when any assertion fails.
|
||||
|
||||
Currently supported assertions:
|
||||
|
||||
- ``bloom_send_rate``: per-node trailing-window ceiling on
|
||||
``stats.bloom.sent`` delta. Calibrated for the bloom-storm
|
||||
regression scenario but generally usable.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .control import snapshot_all_bloom
|
||||
from .scenario import BloomSendRateAssertion, MinParentSwitchesAssertion
|
||||
from .topology import SimTopology
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class AssertionOutcome:
|
||||
name: str
|
||||
passed: bool
|
||||
detail: str
|
||||
|
||||
|
||||
def _bloom_sent_total(node_data: dict) -> int | None:
|
||||
"""Extract stats.bloom.sent from a show_bloom response."""
|
||||
stats = node_data.get("stats") or {}
|
||||
sent = stats.get("sent")
|
||||
if sent is None:
|
||||
return None
|
||||
try:
|
||||
return int(sent)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
class BloomSendRateMonitor:
|
||||
"""Samples per-node ``stats.bloom.sent`` to evaluate a trailing-window
|
||||
ceiling assertion at end-of-run.
|
||||
|
||||
Usage:
|
||||
m = BloomSendRateMonitor(topology, cfg)
|
||||
m.sample_window_start() # called window_secs before scenario end
|
||||
...
|
||||
m.sample_end() # called at scenario end
|
||||
outcome = m.evaluate()
|
||||
"""
|
||||
|
||||
def __init__(self, topology: SimTopology, cfg: BloomSendRateAssertion):
|
||||
self.topology = topology
|
||||
self.cfg = cfg
|
||||
self.window_start: dict[str, int] = {}
|
||||
self.window_end: dict[str, int] = {}
|
||||
|
||||
def sample_window_start(self) -> None:
|
||||
snap = snapshot_all_bloom(self.topology)
|
||||
for nid, data in snap.items():
|
||||
v = _bloom_sent_total(data)
|
||||
if v is not None:
|
||||
self.window_start[nid] = v
|
||||
|
||||
def sample_end(self) -> None:
|
||||
snap = snapshot_all_bloom(self.topology)
|
||||
for nid, data in snap.items():
|
||||
v = _bloom_sent_total(data)
|
||||
if v is not None:
|
||||
self.window_end[nid] = v
|
||||
|
||||
def evaluate(self) -> AssertionOutcome:
|
||||
max_per_node = self.cfg.max_per_node
|
||||
window_secs = self.cfg.window_secs
|
||||
|
||||
if not self.window_start or not self.window_end:
|
||||
return AssertionOutcome(
|
||||
name="bloom_send_rate",
|
||||
passed=False,
|
||||
detail=(
|
||||
f"FAIL bloom_send_rate: failed to sample window endpoints "
|
||||
f"(start={len(self.window_start)} nodes, "
|
||||
f"end={len(self.window_end)} nodes)"
|
||||
),
|
||||
)
|
||||
|
||||
per_node_deltas: dict[str, int] = {}
|
||||
for nid, end_v in self.window_end.items():
|
||||
start_v = self.window_start.get(nid)
|
||||
if start_v is None:
|
||||
continue
|
||||
per_node_deltas[nid] = end_v - start_v
|
||||
|
||||
offenders = {
|
||||
nid: d for nid, d in per_node_deltas.items() if d > max_per_node
|
||||
}
|
||||
max_obs = max(per_node_deltas.values()) if per_node_deltas else 0
|
||||
|
||||
if offenders:
|
||||
sorted_off = sorted(offenders.items(), key=lambda kv: -kv[1])
|
||||
details = ", ".join(f"{nid}={d}" for nid, d in sorted_off)
|
||||
detail = (
|
||||
f"FAIL bloom_send_rate: {len(offenders)} node(s) exceeded "
|
||||
f"ceiling of {max_per_node} bloom_sent over trailing "
|
||||
f"{window_secs}s — offenders: {details} "
|
||||
f"(all per-node deltas: "
|
||||
f"{', '.join(f'{n}={v}' for n, v in sorted(per_node_deltas.items()))})"
|
||||
)
|
||||
return AssertionOutcome(
|
||||
name="bloom_send_rate",
|
||||
passed=False,
|
||||
detail=detail,
|
||||
)
|
||||
|
||||
detail = (
|
||||
f"PASS bloom_send_rate: max per-node delta {max_obs} <= "
|
||||
f"ceiling {max_per_node} over trailing {window_secs}s "
|
||||
f"(per-node: "
|
||||
f"{', '.join(f'{n}={v}' for n, v in sorted(per_node_deltas.items()))})"
|
||||
)
|
||||
return AssertionOutcome(
|
||||
name="bloom_send_rate",
|
||||
passed=True,
|
||||
detail=detail,
|
||||
)
|
||||
|
||||
|
||||
def evaluate_min_parent_switches(
|
||||
cfg: MinParentSwitchesAssertion,
|
||||
parent_switch_count: int,
|
||||
) -> AssertionOutcome:
|
||||
"""Sanity guard: fail the scenario if the harness-induced flap did
|
||||
not produce at least ``cfg.min_total`` parent switches across the
|
||||
run. Detects misconfiguration (e.g., wrong root election) where
|
||||
the bloom-rate assertion would otherwise trivially pass on any
|
||||
binary including the regressed one.
|
||||
"""
|
||||
if parent_switch_count >= cfg.min_total:
|
||||
return AssertionOutcome(
|
||||
name="min_parent_switches",
|
||||
passed=True,
|
||||
detail=(
|
||||
f"PASS min_parent_switches: {parent_switch_count} switches "
|
||||
f">= floor {cfg.min_total}"
|
||||
),
|
||||
)
|
||||
return AssertionOutcome(
|
||||
name="min_parent_switches",
|
||||
passed=False,
|
||||
detail=(
|
||||
f"FAIL min_parent_switches: {parent_switch_count} switches "
|
||||
f"< floor {cfg.min_total} — harness did not induce sufficient "
|
||||
f"parent flapping; bloom-rate assertion would be trivially "
|
||||
f"true. Check tree-snapshot-warmup.json: did the expected "
|
||||
f"node win the root election?"
|
||||
),
|
||||
)
|
||||
@@ -149,6 +149,27 @@ def snapshot_all_mmp(topology: SimTopology) -> dict[str, dict]:
|
||||
return result
|
||||
|
||||
|
||||
def query_bloom(container: str) -> dict | None:
|
||||
"""Query a node's bloom filter state and stats."""
|
||||
return query_node(container, "show_bloom")
|
||||
|
||||
|
||||
def snapshot_all_bloom(topology: SimTopology) -> dict[str, dict]:
|
||||
"""Query show_bloom on all nodes, return {node_id: bloom_data}.
|
||||
|
||||
Nodes that fail to respond are omitted from the result.
|
||||
"""
|
||||
result = {}
|
||||
for node_id in sorted(topology.nodes):
|
||||
container = topology.container_name(node_id)
|
||||
data = query_bloom(container)
|
||||
if data is not None:
|
||||
result[node_id] = data
|
||||
else:
|
||||
log.warning("No bloom data from %s", node_id)
|
||||
return result
|
||||
|
||||
|
||||
def query_routing(container: str) -> dict | None:
|
||||
"""Query a node's routing stats (includes congestion counters)."""
|
||||
return query_node(container, "show_routing")
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Deterministic asymmetric link-cost flapping.
|
||||
|
||||
Periodically rotates a fixed assignment of netem policies across a
|
||||
fixed set of edges. Unlike ``LinkManager`` (random link-down events)
|
||||
or the ``netem.mutation`` block (random per-edge policy mutation),
|
||||
this is a deterministic, periodic flip between named policies on
|
||||
named edges — useful for forcing a downstream node to switch parents
|
||||
on a fixed cadence.
|
||||
|
||||
Rotation is a simple one-position cyclic shift: the policy assigned
|
||||
to edges[0] moves to edges[1], edges[1] -> edges[2], ..., edges[N-1]
|
||||
-> edges[0]. With two edges this degenerates to a swap, which is the
|
||||
intended use for the bloom-storm scenario.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
|
||||
from .netem import NetemManager, NetemParams
|
||||
from .scenario import LinkSwapConfig, NetemPolicy
|
||||
from .topology import SimTopology
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _canonical_edge(edge_str: str) -> tuple[str, str]:
|
||||
"""Parse "nXX-nYY" into a canonical (a, b) tuple sorted alphabetically."""
|
||||
parts = edge_str.split("-")
|
||||
if len(parts) != 2:
|
||||
raise ValueError(f"link_swap edge '{edge_str}' is not 'nXX-nYY' form")
|
||||
a, b = sorted(parts)
|
||||
return a, b
|
||||
|
||||
|
||||
class LinkSwapManager:
|
||||
"""Manages deterministic netem-policy rotation across a fixed edge set."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
topology: SimTopology,
|
||||
config: LinkSwapConfig,
|
||||
netem_mgr: NetemManager,
|
||||
rng: random.Random,
|
||||
):
|
||||
self.topology = topology
|
||||
self.config = config
|
||||
self.netem_mgr = netem_mgr
|
||||
self.rng = rng
|
||||
|
||||
# Resolve edges and validate they exist in the topology
|
||||
topo_edges = {tuple(sorted([a, b])) for a, b in topology.edges}
|
||||
self._edges: list[tuple[str, str]] = []
|
||||
for entry in config.edges:
|
||||
edge = _canonical_edge(entry.edge)
|
||||
if edge not in topo_edges:
|
||||
raise ValueError(
|
||||
f"link_swap edge {entry.edge} not present in topology"
|
||||
)
|
||||
self._edges.append(edge)
|
||||
|
||||
# Current per-position policy name. Index i in this list is
|
||||
# the policy currently applied to self._edges[i]. Each rotate()
|
||||
# cyclically shifts these by one position.
|
||||
self._current_policies: list[str] = [e.policy for e in config.edges]
|
||||
|
||||
self.swap_count = 0
|
||||
# Last apply timestamp; the runner uses this together with
|
||||
# config.interval_secs to schedule rotations.
|
||||
self.last_swap_at: float | None = None
|
||||
|
||||
def policies(self) -> dict[str, NetemPolicy]:
|
||||
return self.config.policies
|
||||
|
||||
def setup_initial(self) -> None:
|
||||
"""Apply the initial policy assignment to each edge.
|
||||
|
||||
Called once after NetemManager.setup_initial() so the link's
|
||||
per-direction tc class state already exists. The initial
|
||||
assignment is taken straight from the config (no rotation).
|
||||
"""
|
||||
for edge, policy_name in zip(self._edges, self._current_policies):
|
||||
policy = self.config.policies[policy_name]
|
||||
self._apply(edge, policy)
|
||||
self.last_swap_at = time.time()
|
||||
log.info(
|
||||
"Link swap initialized: %d edges, interval %.1fs, policies=%s",
|
||||
len(self._edges),
|
||||
self.config.interval_secs,
|
||||
list(self.config.policies.keys()),
|
||||
)
|
||||
|
||||
def maybe_swap(self, now: float) -> bool:
|
||||
"""If the swap interval has elapsed, rotate policies. Return whether
|
||||
a swap occurred so the runner can reschedule.
|
||||
"""
|
||||
if self.last_swap_at is None:
|
||||
self.last_swap_at = now
|
||||
return False
|
||||
if (now - self.last_swap_at) < self.config.interval_secs:
|
||||
return False
|
||||
|
||||
# Cyclic shift by one position
|
||||
self._current_policies = (
|
||||
[self._current_policies[-1]] + self._current_policies[:-1]
|
||||
)
|
||||
for edge, policy_name in zip(self._edges, self._current_policies):
|
||||
policy = self.config.policies[policy_name]
|
||||
self._apply(edge, policy)
|
||||
self.swap_count += 1
|
||||
self.last_swap_at = now
|
||||
log.debug(
|
||||
"Link swap #%d: %s",
|
||||
self.swap_count,
|
||||
", ".join(
|
||||
f"{a}-{b}={p}"
|
||||
for (a, b), p in zip(self._edges, self._current_policies)
|
||||
),
|
||||
)
|
||||
return True
|
||||
|
||||
def _apply(self, edge: tuple[str, str], policy: NetemPolicy) -> None:
|
||||
"""Apply a policy to both directions of an edge using NetemManager."""
|
||||
params = self._sample_policy(policy)
|
||||
# Drive both directions through NetemManager's per-link state so
|
||||
# the rotation persists across mutate() runs and survives node
|
||||
# restarts via setup_node().
|
||||
self.netem_mgr._update_link(edge[0], edge[1], params)
|
||||
|
||||
def _sample_policy(self, policy: NetemPolicy) -> NetemParams:
|
||||
"""Sample concrete params from a policy's ranges (deterministic
|
||||
when min == max, which is the bloom-storm scenario's contract).
|
||||
"""
|
||||
return NetemParams(
|
||||
delay_ms=int(self.rng.uniform(policy.delay_ms[0], policy.delay_ms[1])),
|
||||
jitter_ms=int(self.rng.uniform(policy.jitter_ms[0], policy.jitter_ms[1])),
|
||||
loss_pct=round(
|
||||
self.rng.uniform(policy.loss_pct[0], policy.loss_pct[1]), 1
|
||||
),
|
||||
duplicate_pct=round(
|
||||
self.rng.uniform(policy.duplicate_pct[0], policy.duplicate_pct[1]), 1
|
||||
),
|
||||
reorder_pct=round(
|
||||
self.rng.uniform(policy.reorder_pct[0], policy.reorder_pct[1]), 1
|
||||
),
|
||||
corrupt_pct=round(
|
||||
self.rng.uniform(policy.corrupt_pct[0], policy.corrupt_pct[1]), 1
|
||||
),
|
||||
)
|
||||
@@ -12,10 +12,12 @@ import sys
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
from .assertions import AssertionOutcome, BloomSendRateMonitor, evaluate_min_parent_switches
|
||||
from .compose import generate_compose
|
||||
from .config_gen import write_configs
|
||||
from .control import snapshot_all_congestion, snapshot_all_mmp, snapshot_all_trees
|
||||
from .docker_exec import docker_compose
|
||||
from .link_swap import LinkSwapManager
|
||||
from .links import LinkManager
|
||||
from .logs import AnalysisResult, analyze_logs, collect_logs, write_sim_metadata
|
||||
from .netem import NetemManager
|
||||
@@ -46,10 +48,15 @@ class SimRunner:
|
||||
self.veth_mgr: VethManager | None = None
|
||||
self.netem_mgr: NetemManager | None = None
|
||||
self.link_mgr: LinkManager | None = None
|
||||
self.link_swap_mgr: LinkSwapManager | None = None
|
||||
self.traffic_mgr: TrafficManager | None = None
|
||||
self.node_mgr: NodeManager | None = None
|
||||
self.peer_churn_mgr: PeerChurnManager | None = None
|
||||
|
||||
# Post-run assertion monitors (sampled near end of run).
|
||||
self.bloom_rate_monitor: BloomSendRateMonitor | None = None
|
||||
self.assertion_outcomes: list[AssertionOutcome] = []
|
||||
|
||||
@staticmethod
|
||||
def _resolve_output_dir(scenario: Scenario) -> str:
|
||||
"""Build a timestamped output directory path.
|
||||
@@ -190,6 +197,20 @@ class SimRunner:
|
||||
self.topology, s.link_flaps, self.rng, netem_mgr=self.netem_mgr
|
||||
)
|
||||
|
||||
if s.link_swap.enabled:
|
||||
if not self.netem_mgr:
|
||||
raise RuntimeError(
|
||||
"link_swap requires netem.enabled (depends on per-link tc state)"
|
||||
)
|
||||
self.link_swap_mgr = LinkSwapManager(
|
||||
self.topology, s.link_swap, self.netem_mgr, self.rng,
|
||||
)
|
||||
|
||||
if s.assertions.bloom_send_rate is not None:
|
||||
self.bloom_rate_monitor = BloomSendRateMonitor(
|
||||
self.topology, s.assertions.bloom_send_rate,
|
||||
)
|
||||
|
||||
if s.traffic.enabled:
|
||||
self.traffic_mgr = TrafficManager(
|
||||
self.topology, s.traffic, self.rng, down_nodes=self._down_nodes
|
||||
@@ -226,6 +247,12 @@ class SimRunner:
|
||||
if self.peer_churn_mgr:
|
||||
self.peer_churn_mgr.refresh_all_npubs()
|
||||
|
||||
# Initial link swap policy assignment (after warmup so the netem
|
||||
# tc state is fully set up and the daemons have already
|
||||
# discovered each other under the calm baseline).
|
||||
if self.link_swap_mgr:
|
||||
self.link_swap_mgr.setup_initial()
|
||||
|
||||
def _handle_node_restart(self, node_id: str):
|
||||
"""Called after a node container is restarted.
|
||||
|
||||
@@ -260,12 +287,38 @@ class SimRunner:
|
||||
next_churn = self._schedule_next(start, s.node_churn.interval_secs) if self.node_mgr else float("inf")
|
||||
next_peer_churn = self._schedule_next(start, s.peer_churn.interval_secs) if self.peer_churn_mgr else float("inf")
|
||||
|
||||
# Bloom-send-rate assertion: sample at window_secs before end.
|
||||
bloom_window_start_at = float("inf")
|
||||
if self.bloom_rate_monitor is not None:
|
||||
bloom_window_start_at = (
|
||||
start + duration - s.assertions.bloom_send_rate.window_secs
|
||||
)
|
||||
bloom_window_started = False
|
||||
|
||||
while not self._interrupted:
|
||||
now = time.time()
|
||||
elapsed = now - start
|
||||
if elapsed >= duration:
|
||||
break
|
||||
|
||||
# Bloom-rate window-start sampling
|
||||
if (
|
||||
self.bloom_rate_monitor is not None
|
||||
and not bloom_window_started
|
||||
and now >= bloom_window_start_at
|
||||
):
|
||||
log.info(
|
||||
"Sampling bloom-rate window start (last %ds of run)...",
|
||||
s.assertions.bloom_send_rate.window_secs,
|
||||
)
|
||||
self.bloom_rate_monitor.sample_window_start()
|
||||
bloom_window_started = True
|
||||
|
||||
# Deterministic link swap (before netem mutation so a
|
||||
# mutation round can't clobber the swap mid-tick).
|
||||
if self.link_swap_mgr:
|
||||
self.link_swap_mgr.maybe_swap(now)
|
||||
|
||||
# Netem mutation
|
||||
if self.netem_mgr and now >= next_netem:
|
||||
self.netem_mgr.mutate()
|
||||
@@ -304,6 +357,8 @@ class SimRunner:
|
||||
active = self.traffic_mgr.active_count if self.traffic_mgr else 0
|
||||
peer_churns = self.peer_churn_mgr.churn_count if self.peer_churn_mgr else 0
|
||||
status_extra = f" peer_churns={peer_churns}" if self.peer_churn_mgr else ""
|
||||
if self.link_swap_mgr:
|
||||
status_extra += f" swaps={self.link_swap_mgr.swap_count}"
|
||||
print(
|
||||
f"\r [{elapsed:.0f}s/{duration}s] "
|
||||
f"nodes={len(self.topology.nodes)} "
|
||||
@@ -320,11 +375,49 @@ class SimRunner:
|
||||
|
||||
print() # Clear status line
|
||||
|
||||
# Bloom-rate window-end sampling.
|
||||
# Done before teardown so containers are still running.
|
||||
if self.bloom_rate_monitor is not None:
|
||||
if not bloom_window_started:
|
||||
# Loop exited before the window-start mark (e.g.,
|
||||
# interrupted). Take both samples now so we still
|
||||
# produce a finite outcome rather than an empty dict.
|
||||
log.warning(
|
||||
"Bloom-rate window did not start during run; "
|
||||
"sampling both endpoints at end (delta will be 0)."
|
||||
)
|
||||
self.bloom_rate_monitor.sample_window_start()
|
||||
log.info("Sampling bloom-rate window end...")
|
||||
self.bloom_rate_monitor.sample_end()
|
||||
|
||||
def _evaluate_assertions(self) -> None:
|
||||
"""Evaluate post-run assertions and stash outcomes on self.
|
||||
|
||||
Called from teardown while containers are still running so the
|
||||
assertion outcomes (which include per-node detail) can be
|
||||
written alongside the run artifacts.
|
||||
"""
|
||||
if self.bloom_rate_monitor is not None:
|
||||
outcome = self.bloom_rate_monitor.evaluate()
|
||||
self.assertion_outcomes.append(outcome)
|
||||
if outcome.passed:
|
||||
log.info("%s", outcome.detail)
|
||||
else:
|
||||
log.error("%s", outcome.detail)
|
||||
|
||||
@property
|
||||
def assertions_failed(self) -> bool:
|
||||
return any(not o.passed for o in self.assertion_outcomes)
|
||||
|
||||
def _teardown(self) -> AnalysisResult | None:
|
||||
"""Stop dynamic elements, collect logs, analyze, stop containers."""
|
||||
result = None
|
||||
|
||||
if self.topology and self.compose_file:
|
||||
# Evaluate post-run assertions before doing any teardown so
|
||||
# control sockets are still reachable.
|
||||
self._evaluate_assertions()
|
||||
|
||||
# Stop traffic
|
||||
if self.traffic_mgr:
|
||||
log.info("Stopping traffic sessions...")
|
||||
@@ -366,6 +459,30 @@ class SimRunner:
|
||||
f.write(result.summary())
|
||||
print(result.summary())
|
||||
|
||||
# Log-derived assertions (evaluated after analyze_logs so
|
||||
# parent_switches and similar are populated).
|
||||
mps_cfg = self.scenario.assertions.min_parent_switches
|
||||
if mps_cfg is not None:
|
||||
outcome = evaluate_min_parent_switches(
|
||||
mps_cfg, len(result.parent_switches)
|
||||
)
|
||||
self.assertion_outcomes.append(outcome)
|
||||
if outcome.passed:
|
||||
log.info("%s", outcome.detail)
|
||||
else:
|
||||
log.error("%s", outcome.detail)
|
||||
|
||||
# Write assertion outcomes
|
||||
if self.assertion_outcomes:
|
||||
assertions_path = os.path.join(self.output_dir, "assertions.txt")
|
||||
with open(assertions_path, "w") as f:
|
||||
for o in self.assertion_outcomes:
|
||||
f.write(o.detail + "\n")
|
||||
print("=== Assertions ===")
|
||||
for o in self.assertion_outcomes:
|
||||
print(o.detail)
|
||||
print()
|
||||
|
||||
# Write metadata
|
||||
write_sim_metadata(
|
||||
self.output_dir,
|
||||
|
||||
@@ -150,6 +150,67 @@ class IngressConfig:
|
||||
burst_bytes: int = 32000
|
||||
|
||||
|
||||
@dataclass
|
||||
class LinkSwapEdge:
|
||||
"""One edge in a link-swap rotation.
|
||||
|
||||
Edge is a canonical "nXX-nYY" string. ``policy`` names a policy
|
||||
in ``LinkSwapConfig.policies``.
|
||||
"""
|
||||
|
||||
edge: str = ""
|
||||
policy: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class LinkSwapConfig:
|
||||
"""Deterministic asymmetric link-cost flapping.
|
||||
|
||||
On each ``interval_secs``, the policies on every pair of edges in
|
||||
``edges`` are swapped (cyclically rotated by one position). This
|
||||
differs from ``link_flaps`` (random link-down events) and
|
||||
``netem.mutation`` (random per-edge policy mutation) by being a
|
||||
deterministic, periodic flip between two named netem policies on
|
||||
a fixed set of edges.
|
||||
|
||||
Used to drive a downstream node to repeatedly switch parents on
|
||||
a fixed cadence — exercises the spanning-tree rebalance path
|
||||
without the noise of a random mutation walk.
|
||||
"""
|
||||
|
||||
enabled: bool = False
|
||||
interval_secs: float = 4.0
|
||||
policies: dict[str, NetemPolicy] = field(default_factory=dict)
|
||||
edges: list[LinkSwapEdge] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class BloomSendRateAssertion:
|
||||
"""Trailing-window ceiling on per-node ``stats.bloom.sent`` delta."""
|
||||
|
||||
window_secs: int = 30
|
||||
max_per_node: int = 30
|
||||
|
||||
|
||||
@dataclass
|
||||
class MinParentSwitchesAssertion:
|
||||
"""Sanity guard: total parent switches across the run must be at
|
||||
least ``min_total``. Used to detect a misconfigured harness where
|
||||
the flap inducer is firing but the topology never produces a real
|
||||
parent-switch event (e.g., wrong root election).
|
||||
"""
|
||||
|
||||
min_total: int = 1
|
||||
|
||||
|
||||
@dataclass
|
||||
class AssertionsConfig:
|
||||
"""Optional post-run assertions evaluated against control-socket data."""
|
||||
|
||||
bloom_send_rate: BloomSendRateAssertion | None = None
|
||||
min_parent_switches: MinParentSwitchesAssertion | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class LoggingConfig:
|
||||
rust_log: str = "info"
|
||||
@@ -169,6 +230,8 @@ class Scenario:
|
||||
peer_churn: PeerChurnConfig = field(default_factory=PeerChurnConfig)
|
||||
bandwidth: BandwidthConfig = field(default_factory=BandwidthConfig)
|
||||
ingress: IngressConfig = field(default_factory=IngressConfig)
|
||||
link_swap: LinkSwapConfig = field(default_factory=LinkSwapConfig)
|
||||
assertions: AssertionsConfig = field(default_factory=AssertionsConfig)
|
||||
logging: LoggingConfig = field(default_factory=LoggingConfig)
|
||||
# Raw YAML dict appended to each generated FIPS node config.
|
||||
# Allows scenarios to override any FIPS config parameter
|
||||
@@ -320,6 +383,40 @@ def load_scenario(path: str) -> Scenario:
|
||||
s.ingress.tiers_kbps = [int(t) for t in tiers]
|
||||
s.ingress.burst_bytes = int(ig.get("burst_bytes", 32000))
|
||||
|
||||
# Link swap section (deterministic asymmetric link-cost flapping).
|
||||
ls = raw.get("link_swap", {})
|
||||
s.link_swap.enabled = ls.get("enabled", False)
|
||||
if "interval_secs" in ls:
|
||||
s.link_swap.interval_secs = float(ls["interval_secs"])
|
||||
if "policies" in ls:
|
||||
s.link_swap.policies = {
|
||||
name: _parse_netem_policy(pdata)
|
||||
for name, pdata in ls["policies"].items()
|
||||
}
|
||||
if "edges" in ls:
|
||||
for edata in ls["edges"]:
|
||||
if not isinstance(edata, dict):
|
||||
raise ValueError("link_swap.edges entries must be dicts")
|
||||
edge = str(edata.get("edge", ""))
|
||||
policy = str(edata.get("policy", ""))
|
||||
if not edge or not policy:
|
||||
raise ValueError("link_swap.edges entries require 'edge' and 'policy'")
|
||||
s.link_swap.edges.append(LinkSwapEdge(edge=edge, policy=policy))
|
||||
|
||||
# Assertions section (post-run control-socket-based checks).
|
||||
asrt = raw.get("assertions", {})
|
||||
if "bloom_send_rate" in asrt:
|
||||
bsr = asrt["bloom_send_rate"]
|
||||
s.assertions.bloom_send_rate = BloomSendRateAssertion(
|
||||
window_secs=int(bsr.get("window_secs", 30)),
|
||||
max_per_node=int(bsr.get("max_per_node", 30)),
|
||||
)
|
||||
if "min_parent_switches" in asrt:
|
||||
mps = asrt["min_parent_switches"]
|
||||
s.assertions.min_parent_switches = MinParentSwitchesAssertion(
|
||||
min_total=int(mps.get("min_total", 1)),
|
||||
)
|
||||
|
||||
# Logging section
|
||||
lg = raw.get("logging", {})
|
||||
s.logging.rust_log = lg.get("rust_log", "info")
|
||||
@@ -435,3 +532,29 @@ def _validate(s: Scenario):
|
||||
raise ValueError(f"ingress.tiers_kbps: all values must be > 0, got {tier}")
|
||||
if s.ingress.burst_bytes <= 0:
|
||||
raise ValueError(f"ingress.burst_bytes must be > 0, got {s.ingress.burst_bytes}")
|
||||
|
||||
# Validate link_swap
|
||||
if s.link_swap.enabled:
|
||||
if s.link_swap.interval_secs <= 0:
|
||||
raise ValueError("link_swap.interval_secs must be > 0")
|
||||
if len(s.link_swap.edges) < 2:
|
||||
raise ValueError("link_swap.edges must list at least 2 edges to swap")
|
||||
if not s.link_swap.policies:
|
||||
raise ValueError("link_swap.policies must not be empty when link_swap.enabled")
|
||||
for entry in s.link_swap.edges:
|
||||
if entry.policy not in s.link_swap.policies:
|
||||
raise ValueError(
|
||||
f"link_swap.edges: policy '{entry.policy}' not in link_swap.policies"
|
||||
)
|
||||
|
||||
# Validate assertions
|
||||
if s.assertions.bloom_send_rate is not None:
|
||||
bsr = s.assertions.bloom_send_rate
|
||||
if bsr.window_secs < 1:
|
||||
raise ValueError("assertions.bloom_send_rate.window_secs must be >= 1")
|
||||
if bsr.max_per_node < 0:
|
||||
raise ValueError("assertions.bloom_send_rate.max_per_node must be >= 0")
|
||||
if bsr.window_secs > s.duration_secs:
|
||||
raise ValueError(
|
||||
"assertions.bloom_send_rate.window_secs must not exceed scenario duration"
|
||||
)
|
||||
|
||||
@@ -23,6 +23,7 @@
|
||||
# chaos-ethernet-only, chaos-tcp-mesh, chaos-bottleneck-parent,
|
||||
# chaos-cost-avoidance, chaos-cost-reeval, chaos-cost-stability,
|
||||
# chaos-depth-vs-cost, chaos-mixed-technology, chaos-congestion-stress,
|
||||
# chaos-bloom-storm,
|
||||
# sidecar, dns-resolver, deb-install
|
||||
#
|
||||
# Opt-in (require --with-tor; depend on live Tor network):
|
||||
@@ -70,6 +71,7 @@ CHAOS_SUITES=(
|
||||
"depth-vs-cost depth-vs-cost"
|
||||
"mixed-technology mixed-technology"
|
||||
"congestion-stress congestion-stress"
|
||||
"bloom-storm bloom-storm"
|
||||
)
|
||||
GATEWAY_SUITES=(gateway)
|
||||
SIDECAR_SUITES=(sidecar)
|
||||
|
||||
Reference in New Issue
Block a user