je_load_density.utils.reliability.network_conditioner

Network conditioner.

Per-task latency / jitter / packet-loss injector implemented inside the scenario runner, so it works against every user template uniformly (no kernel tc required). Drops are simulated by raising a ConnectionError before the request fires.

Functions

`current_conditioner`()
`install_network_conditioner`([latency_ms, ...])
`uninstall_network_conditioner`()

Classes

NetworkConditioner([latency_ms, jitter_ms, ...])

class je_load_density.utils.reliability.network_conditioner.NetworkConditioner(latency_ms: float = 0.0, jitter_ms: float = 0.0, loss_rate: float = 0.0, name_filter: str | None = None)

Bases: object

applies_to(task: Dict[str, Any]) → bool

apply(task: ~typing.Dict[str, ~typing.Any], sleeper: ~typing.Callable[[float], None] = <built-in function sleep>) → None

jitter_ms: float = 0.0

latency_ms: float = 0.0

loss_rate: float = 0.0

name_filter: str | None = None

je_load_density.utils.reliability.network_conditioner.current_conditioner() → NetworkConditioner | None

je_load_density.utils.reliability.network_conditioner.install_network_conditioner(latency_ms: float = 0.0, jitter_ms: float = 0.0, loss_rate: float = 0.0, name_filter: str | None = None) → NetworkConditioner

je_load_density.utils.reliability.network_conditioner.uninstall_network_conditioner() → None