Network Emulation Service#

Introduction#

This module exposes different ways of doing network emulation [1] based on tc [2] of the iproute2 tool and use netem [3] as cornerstone.

Different kinds of network topologies can be built and different level of abstractions can be used.

At the network device interface level (the lowest level for now) you can use netem() to set inbound and/or outbound limitation on arbitrary network interfaces of your nodes. Nodes can be seen as the vertices of a star topology where the center is the core of the network. For instance the higher the latency on a node is, the further it is from the core of the network.

The above prevent you to set heterogeneous limitations ie based on the packets destinations. The function netem_htb() enforces Hierarchical Token Bucket [4] on your nodes. This lets you modeling a n-simplex topology where the vertices represent your nodes and the edges the limitations.

Working at the device interface level provides you with great flexibility, however this comes at the cost of being very explicit. For instance, you must know in advance the device names, the IP target… From a higher perspective three Services, working at the role level are provided. The services will infer everything for you (device names, target IPs) based on high level parameters. Of course, those services are using internally the above functions.

Netem allows for emulating a star topology: you control the constraints of a node to/from the center of the topology.
NetemHTB will enforce heterogeneous network limitations based on the role names (and thus based on the packets’ destination)
AccurateNetemHTB inherits from the previous one, but will fix the latency limitations so that the user defined delays match the end-to-end latency of the application.

Note

Requirements:

ifb module loaded on the remote hosts for Netem service and netem function.

tc tool available (install iproute2 package on debian based distribution)

Netem emulation#

Classes:

`Netem`()	Set homogeneous network constraints on some hosts
`NetemInConstraint`(device, options)	A Constraint on the ingress part of a device.
`NetemInOutSource`(host, constraints)	Model a host and the constraints on its network devices.
`NetemOutConstraint`(device, options)	A Constraint on the egress part of a device.

Functions:

netem(sources[, chunk_size])

Helper function to enforce in/out limitations on host devices.

class enoslib.service.emul.netem.Netem#

Set homogeneous network constraints on some hosts

Geometrically speaking: nodes are put on a star topology. Limitation are set from a node to/from the center of the Network.

This calls netem() function internally. As a consequence when symmetric is True, it will apply 4 times the constraints for bidirectional communication (out_a, in_b, out_b, in_a) and 2 times if symmetric is False.

Example

import logging

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

conf = (
    en.G5kConf.from_settings(job_type=[], walltime="01:00:00")
    .add_machine(
        roles=["city", "paris"],
        cluster="paravance",
        nodes=1,
    )
    .add_machine(
        roles=["city", "berlin"],
        cluster="paravance",
        nodes=1,
    )
    .add_machine(
        roles=["city", "londres"],
        cluster="paravance",
        nodes=1,
    )
)
provider = en.G5k(conf)
roles, networks = provider.init()
roles = en.sync_info(roles, networks)

netem = en.Netem()
(
    netem.add_constraints("delay 10ms", roles["paris"], symmetric=True)
    .add_constraints("delay 20ms", roles["londres"], symmetric=True)
    .add_constraints("delay 30ms", roles["berlin"], symmetric=True)
)

netem.deploy()
netem.validate()
netem.destroy()

for role, hosts in roles.items():
    print(role)
    for host in hosts:
        print(f"-- {host.alias}")

Using a secondary network.

import logging
from pathlib import Path

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

CLUSTER = "paravance"
SITE = en.g5k_api_utils.get_cluster_site(CLUSTER)

job_name = Path(__file__).name

private = en.G5kNetworkConf(type="kavlan-global", roles=["private"], site=SITE)

conf = (
    en.G5kConf.from_settings(
        job_name=job_name,
        walltime="00:30:00",
        job_type=["deploy"],
        env_name="debian11-nfs",
    )
    .add_network_conf(private)
    .add_machine(
        roles=["paris"],
        cluster=CLUSTER,
        nodes=1,
        secondary_networks=[private],
    )
    .add_machine(
        roles=["londres"],
        cluster=CLUSTER,
        nodes=1,
        secondary_networks=[private],
    )
    .add_machine(
        roles=["berlin"],
        cluster=CLUSTER,
        nodes=1,
        secondary_networks=[private],
    )
)

provider = en.G5k(conf)
roles, networks = provider.init()
roles = en.sync_info(roles, networks)

netem = en.Netem()
(
    netem.add_constraints(
        "delay 10ms", roles["paris"], networks=networks["private"], symmetric=True
    )
    .add_constraints(
        "delay 20ms", roles["londres"], networks=networks["private"], symmetric=True
    )
    .add_constraints(
        "delay 30ms", roles["berlin"], networks=networks["private"], symmetric=True
    )
)

netem.deploy()
netem.validate()

for role, hosts in roles.items():
    print(role)
    for host in hosts:
        print(f"-- {host.alias}")

backup()#: (abstract) Backup the service.

deploy(chunk_size=100, **kwargs)#: Apply the constraints on all the hosts.

destroy(**kwargs)#: (abstract) Destroy the service.

class enoslib.service.emul.netem.NetemInConstraint(device: str, options: str)#

A Constraint on the ingress part of a device.

Inbound limitations works differently. see https://wiki.linuxfoundation.org/networking/netem

We’ll create an ifb, redirect incoming traffic to it and apply some queuing discipline using netem.

Parameters:: ifb – the ifb name (e.g. ifb0) that will be used. That means that the various ifbs must be provisioned out of band.

add_commands(ifb: str) → List[str]#: Return the commands that adds the ifb.

commands(ifb: str) → List[str]#: Return the commands that redirect and apply netem constraints on the ifb.

remove_commands(ifb: str) → List[str]#: Return the commands that remove the qdisc from the ifb and the net device.

class enoslib.service.emul.netem.NetemInOutSource(host: ~enoslib.objects.Host, constraints: ~typing.Set[~enoslib.service.emul.netem.NetemConstraint] = <factory>)#

Model a host and the constraints on its network devices.

Parameters:

inbound – The constraints to set on the ingress part of the host devices
outbound – The constraints to set on the egress part of the host devices

add_constraints(constraints: Iterable[NetemConstraint])#

Merge constraints to existing ones.

In this context if two constraints are set on the same device we overwrite the original one. At the end we ensure that there’s only one constraint per device

Parameters:: constraints – Iterable of NetemIn[Out]Constraint

equal(c1: NetemConstraint, c2: NetemConstraint) → bool#: Encode the equality between two constraints in this context.

class enoslib.service.emul.netem.NetemOutConstraint(device: str, options: str)#

A Constraint on the egress part of a device.

Parameters:

device – the device name where the qdisc will be added
options – the options string the pass down to netem (e.g. delay 10ms)

commands(_: str) → List[str]#: Nothing to do.

enoslib.service.emul.netem.netem(sources: List[NetemInOutSource], chunk_size: int = 100, **kwargs)#

Helper function to enforce in/out limitations on host devices.

Nodes can be seen as the vertices of a star topology where the center is the core of the network. For instance the higher the latency on a node is, the further it is from the core of the network.

This method is optimized toward execution time: enforcing thousands of atomic constraints (= tc commands) shouldn’t be a problem. Commands are sent by batch and chunk_size controls the size of the batch.

Idempotency note: the existing qdiscs are removed before applying new ones. This must be safe in most of the cases to consider that this is a form of idempotency.

Parameters:

sources – list of constraints to apply as a list of Source
chunk_size – size of the chunk to use
kwargs – keyword argument to pass to enoslib.api.run_ansible().

Example

import logging
from pathlib import Path

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

job_name = Path(__file__).name

conf = (
    en.G5kConf.from_settings(job_name=job_name, job_type=[])
    .add_machine(
        roles=["city", "paris"],
        cluster="paravance",
        nodes=1,
    )
    .add_machine(
        roles=["city", "berlin"],
        cluster="paravance",
        nodes=1,
    )
    .add_machine(
        roles=["city", "londres"],
        cluster="paravance",
        nodes=1,
    )
)
provider = en.G5k(conf)
roles, networks = provider.init()

sources = []
for idx, host in enumerate(roles["city"]):
    delay = 5 * idx
    print(f"{host.alias} <-> {delay}")
    inbound = en.NetemOutConstraint(device="br0", options=f"delay {delay}ms")
    outbound = en.NetemInConstraint(device="br0", options=f"delay {delay}ms")
    sources.append(en.NetemInOutSource(host, constraints={inbound, outbound}))

en.netem(sources)

HTB (Hierarchical Token Bucket) emulation#

HTB based emulation.

Classes:

`AccurateNetemHTB`()	NetemHTB but for expressing end-to-end latency.
`HTBConstraint`(device, delay, target[, rate, ...])	An HTB constraint.
`HTBSource`(host, constraints)	Model a host and all the htb constraints.
`NetemHTB`()	Add extra delay to outgoing packets based on their destination.

Functions:

netem_htb(htb_hosts[, chunk_size])

Helper function to enforce heterogeneous limitations on hosts.

class enoslib.service.emul.htb.AccurateNetemHTB#

NetemHTB but for expressing end-to-end latency.

End-to-End (one way )latency between two distributed processes is composed of:

the latency for a packet to go out of the first machine (e.g. IP stack latency + network card latency).
the latency for the packet to flight to the second machine (e.g. network, middle-boxes latency)
the latency for a packet to be transferred to the application on the second machine.

This service tries the compensated the extra latency added to outgoing packets by TC so that the End-to-End latency stick to the one specified by the user.

Internals hints:

At deploy time AccurateNetemHTB will estimate the delay introduced by the infrastructure and correct the wanted netem by this estimation. For each source and target in the wanted constraints it will send N ICMP probes and aggregate their RTT time. The current estimation is based on the mean RTT.

The AccurateNetemHTB is useful when the delays introduced by the infrastructure aren’t negligible in comparison to the wanted ones (e.g few ms in a LAN environment). Since correction is applied on each pair (source, destination), the service can cope with the heterogeneity of delays in the infrastructure.

Of course this service can’t do any miracle like trying to set delays less than the one existing on the infrastructure. Always double check, the observed network condition before drawing any conclusion.

Add extra delay to outgoing packets based on their destination.

It allows to setup complex network topology. For a much simpler way of applying constraints see Netem

The topology can be built by add constraints iteratively with add_constraints() or by passing a description as a dictionary using the from_dict() class method.

deploy(chunk_size: int = 100, **kwargs)#

Deploy the network emulation.

This is where the hard work is done:

estimate the current status of the network
correct the user defined constraints
enforce them

Parameters:

chunk_size – see deploy()
kwargs – see deploy()

class enoslib.service.emul.htb.HTBConstraint(device: str, delay: str, target: str, rate: str = '10gbit', loss: str | None = None)#

An HTB constraint.

Note

Ref: https://tldp.org/HOWTO/Adv-Routing-HOWTO/lartc.qdisc.classful.html

An HTBconstraint will be enforced by creating a filter and a class. The filtering stage will send the packet to the class (delay, rate and loss will be applied) based on the target ip.

The purpose of this class is to give you the commands to create the slice (class and filter) that will be added to the root qdisc.

Parameters:

target – the target ip Can be an ipv4 or an ipv6.
device – the device name where the qdisc will be added The one you get when listing the interfaces using iproute2.
delay – the delay (e.g 10ms) One way delay.
rate – the rate (e.g 10gbit) Bandwitdth of the link
loss – the loss (e.g “0.05” == “5%”) Chance for a packet to be lost

add_commands() → List[str]#: Add the class-full qdisc at the root of the device.

commands(idx: int)#: Get the command for the current slice.

remove_commands() → List[str]#: Remove everything.

class enoslib.service.emul.htb.HTBSource(host: ~enoslib.objects.Host, constraints: ~typing.Set[~enoslib.service.emul.htb.HTBConstraint] = <factory>)#

Model a host and all the htb constraints.

Args: host: Host where the constraint will be applied constraints: list of enoslib.service.netem.netem.HTBConstraint

Note

Consistency check (such as device names) between the host and the constraints are left to the application.

add_constraint(*args, **kwargs)#

Add a constraint.

*args and **kwargs are those from enoslib.service.netem.netem.HTBConstraint

add_constraints(constraints: Iterable[HTBConstraint]) → HTBSource#

Merge constraints to existing ones.

In this context if two constraints are set on the same device with the same target will overwrite the original one. At the end we ensure that there’s only one constraint per device and target

Parameters:: constraints – Iterable of HTBConstraints

static equal(c1: HTBConstraint, c2: HTBConstraint) → bool#: Encode the equality of two constraints in this context.

class enoslib.service.emul.htb.NetemHTB#

Add extra delay to outgoing packets based on their destination.

It allows to setup complex network topology. For a much simpler way of applying constraints see Netem

The topology can be built by add constraints iteratively with add_constraints() or by passing a description as a dictionary using the from_dict() class method.

add_constraints(src: Iterable[Host], dest: Iterable[Host], delay: str, rate: str, loss: float | None = None, networks: Iterable[Network] | None = None, symmetric: bool = False, *, symetric: bool | None = None) → NetemHTB#

Add some constraints.

Parameters:

src – list of hosts on which the constraint will be applied
dest – list of hosts to which traffic will be limited
delay – the delay to apply as a string (e.g. 10ms)
rate – the rate to apply as a string (e.g. 1gbit)
loss – the percentage of loss (between 0 and 1)
networks – only consider these networks when applying the resources (default to all networks)
symmetric – True iff the symmetric rules should be also added.

Returns:

The current service with updated constraints. (This allows to chain the addition of constraints)

Examples

import logging
from pathlib import Path

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

job_name = Path(__file__).name

conf = (
    en.G5kConf.from_settings(job_name=job_name, job_type=[])
    .add_machine(roles=["paris"], cluster="paravance", nodes=1)
    .add_machine(roles=["berlin"], cluster="paravance", nodes=1)
    .add_machine(roles=["londres"], cluster="paravance", nodes=1)
)
provider = en.G5k(conf)
roles, networks = provider.init()
roles = en.sync_info(roles, networks)


netem = en.NetemHTB()
(
    netem.add_constraints(
        src=roles["paris"],
        dest=roles["londres"],
        delay="10ms",
        rate="1gbit",
        symmetric=True,
    )
    .add_constraints(
        src=roles["paris"],
        dest=roles["berlin"],
        delay="20ms",
        rate="1gbit",
        symmetric=True,
    )
    .add_constraints(
        src=roles["londres"],
        dest=roles["berlin"],
        delay="20ms",
        rate="1gbit",
        symmetric=True,
    )
)
netem.deploy()
netem.validate()
# netem.destroy()

Using a secondary network from a list of constraints

import logging
from pathlib import Path

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

CLUSTER = "paravance"
SITE = en.g5k_api_utils.get_cluster_site(CLUSTER)

job_name = Path(__file__).name

private = en.G5kNetworkConf(type="kavlan", roles=["private"], site=SITE)

conf = (
    en.G5kConf.from_settings(
        job_name=job_name, job_type=["deploy"], env_name="debian11-nfs"
    )
    .add_network_conf(private)
    .add_machine(
        roles=["paris"],
        cluster=CLUSTER,
        nodes=1,
        secondary_networks=[private],
    )
    .add_machine(
        roles=["londres"],
        cluster=CLUSTER,
        nodes=1,
        secondary_networks=[private],
    )
    .add_machine(
        roles=["berlin"],
        cluster=CLUSTER,
        nodes=1,
        secondary_networks=[private],
    )
)

provider = en.G5k(conf)
roles, networks = provider.init()
roles = en.sync_info(roles, networks)


netem = en.NetemHTB()
(
    netem.add_constraints(
        src=roles["paris"],
        dest=roles["londres"],
        delay="10ms",
        rate="1gbit",
        symmetric=True,
        networks=networks["private"],
    )
    .add_constraints(
        src=roles["paris"],
        dest=roles["berlin"],
        delay="20ms",
        rate="1gbit",
        symmetric=True,
        networks=networks["private"],
    )
    .add_constraints(
        src=roles["londres"],
        dest=roles["berlin"],
        delay="20ms",
        rate="1gbit",
        symmetric=True,
        networks=networks["private"],
    )
)
netem.deploy()
netem.validate()
# netem.destroy()

backup()#

(Not Implemented) Backup.

Feel free to share your ideas.

deploy(chunk_size: int = 100, **kwargs) → List[HTBSource]#: (abstract) Deploy the service.

destroy(**kwargs)#

Reset the network constraints(latency, bandwidth …)

Remove any filter that have been applied to shape the traffic ever.

Careful: This remove every rule, including those not managed by this service.

classmethod from_dict(network_constraints: Dict, roles: Roles, networks: Networks) → NetemHTB#

Build the service from a dictionary describing the network topology.

Parameters:: network_constraints – Dictionary of constraints. This must conform with SCHEMA.

Examples

import logging
from pathlib import Path

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

job_name = Path(__file__).name

conf = (
    en.G5kConf.from_settings(job_name=job_name, job_type=[])
    .add_machine(
        roles=["paris", "vm"],
        cluster="paravance",
        nodes=1,
    )
    .add_machine(
        roles=["berlin", "vm"],
        cluster="paravance",
        nodes=1,
    )
    .add_machine(
        roles=["londres", "vm"],
        cluster="paravance",
        nodes=1,
    )
)
provider = en.G5k(conf)
roles, networks = provider.init()
roles = en.sync_info(roles, networks)

# Building the network constraints
emulation_conf = {
    "default_delay": "20ms",
    "default_rate": "1gbit",
    "except": [],
    "constraints": [
        {"src": "paris", "dst": "londres", "symmetric": True, "delay": "10ms"}
    ],
}

logging.info(emulation_conf)

netem = en.NetemHTB.from_dict(emulation_conf, roles, networks)
netem.deploy()
netem.validate()
# netem.destroy()

Using a secondary network:

import logging
from pathlib import Path

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

CLUSTER = "paravance"
SITE = en.g5k_api_utils.get_cluster_site(CLUSTER)

job_name = Path(__file__).name

prod = en.G5kNetworkConf(type="prod", roles=["my_network"], site=SITE)
private = en.G5kNetworkConf(type="kavlan", roles=["private"], site=SITE)

conf = (
    en.G5kConf.from_settings(
        job_name=job_name, job_type=["deploy"], env_name="debian11-nfs"
    )
    .add_network_conf(prod)
    .add_network_conf(private)
    .add_machine(
        roles=["paris"],
        cluster=CLUSTER,
        nodes=1,
        primary_network=prod,
        secondary_networks=[private],
    )
    .add_machine(
        roles=["londres"],
        cluster=CLUSTER,
        nodes=1,
        primary_network=prod,
        secondary_networks=[private],
    )
    .add_machine(
        roles=["berlin"],
        cluster=CLUSTER,
        nodes=1,
        primary_network=prod,
        secondary_networks=[private],
    )
)

provider = en.G5k(conf)
roles, networks = provider.init()
roles = en.sync_info(roles, networks)

# Building the network constraints
emulation_conf = {
    "default_delay": "20ms",
    "default_rate": "1gbit",
    "except": [],
    "default_network": "private",
    "constraints": [
        {"src": "paris", "dst": "londres", "symmetric": True, "delay": "10ms"}
    ],
}

logging.info(emulation_conf)

netem = en.NetemHTB.from_dict(emulation_conf, roles, networks)
netem.destroy()
netem.deploy()
netem.validate()

validate(*, networks: Iterable[Network] | None = None, output_dir: Path | str | None = None, **kwargs) → Results#

Validate the network parameters(latency, bandwidth …)

Performs ping tests to validate the constraints set by deploy(). Reports are available in the putput directory. used by enos.

Parameters:

roles (dict) – role -> hosts mapping as returned by enoslib.infra.provider.Provider.init()
inventory_path (str) – path to an inventory
output_dir (str) – directory where validation files will be stored. Default to enoslib.constants.TMP_DIRNAME.

enoslib.service.emul.htb.netem_htb(htb_hosts: List[HTBSource], chunk_size: int = 100, **kwargs)#

Helper function to enforce heterogeneous limitations on hosts.

This function do the heavy lifting of building the qdisc tree for each node and add filters based on the ip destination of the packet. Ref: https://tldp.org/HOWTO/Traffic-Control-HOWTO/classful-qdiscs.html

This function is used internally by the enoslib.service.netem.netem.Netem service. Here, you must ensure that the various enoslib.service.netem.netem.HTBSource are consistent (e.g. device names.) with the host. Symmetric limitations can be achieved by adding both end of the communication to the list. The same host can appear multiple times in the list, all the constraints will be merged according to enoslib.service.emul.htb.HTBSource.add_constraints()

Idempotency note: the existing qdiscs are removed before applying new ones. This must be safe in most of the cases to consider that this is a form of idempotency.

Parameters:

htb_hosts – list of constraints to apply.
chunk_size – size of the chunk to use
kwargs – keyword arguments passed to enoslib.api.run_ansible()

Examples

import logging
from itertools import islice, product
from pathlib import Path

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()

job_name = Path(__file__).name


conf = (
    en.G5kConf.from_settings(job_name=job_name, job_type=[])
    .add_network(
        id="not_linked_to_any_machine",
        type="slash_22",
        roles=["my_subnet"],
        site="rennes",
    )
    .add_machine(roles=["control"], cluster="paravance", nodes=10)
)

provider = en.G5k(conf)

# Get actual resources
roles, networks = provider.init()

# distribute some virtual ips :)
N = 100
ips = networks["my_subnet"][0].free_ips
for host in roles["control"]:
    host.extra.update(ips=[str(ip) for ip in islice(ips, N)])
with en.play_on(roles=roles, gather_facts=False) as p:
    p.shell(
        "(ip a | grep {{ item }}) || ip addr add {{ item }}/22 dev br0",
        loop="{{ ips }}",
    )

# All virtual ips being set on the hosts
# let's build the list of constraints
htb_hosts = []
humans = []
for h_idx, (h1, h2) in enumerate(product(roles["control"], roles["control"])):
    # need to account for h1 = h2 to set constraint on loopback device
    htb = en.HTBSource(host=h1)
    ips2 = h2.extra["ips"]
    for ip_idx, ip2 in enumerate(ips2):
        # this is the delay between one machine h1 and any of the virtual ip of h2
        # since h1 and h2 will be swapped in another iteration, we'll also set
        # the "symmetrical" at some point.
        delay = 5 * ip_idx
        humans.append(f"({h1.alias}) -->{delay}--> {ip2}({h2.alias}) ")
        if h1 == h2:
            # loopback
            htb.add_constraint(delay=f"{delay}ms", device="lo", target=ip2)
        else:
            htb.add_constraint(delay=f"{delay}ms", device="br0", target=ip2)
    htb_hosts.append(htb)

en.netem_htb(htb_hosts)


Path("htb.list").write_text("\n".join(humans))
# you can check the constraint be issuing some:
# ping -I <ip_source> <ip_dest>