diff options
| -rw-r--r-- | rumba/model.py | 212 | ||||
| -rw-r--r-- | rumba/testbeds/dockertb.py | 5 | ||||
| -rw-r--r-- | rumba/testbeds/emulab.py | 5 | ||||
| -rw-r--r-- | rumba/testbeds/jfed.py | 4 | ||||
| -rw-r--r-- | rumba/testbeds/local.py | 4 | ||||
| -rw-r--r-- | rumba/testbeds/qemu.py | 17 |
6 files changed, 211 insertions, 36 deletions
diff --git a/rumba/model.py b/rumba/model.py index d46cff9..b6cb15e 100644 --- a/rumba/model.py +++ b/rumba/model.py @@ -8,6 +8,7 @@ # Dimitri Staessens <dimitri.staessens@ugent.be> # Vincenzo Maffione <v.maffione@nextworks.it> # Marco Capitani <m.capitani@nextworks.it> +# Nick Aerts <nick.aerts@ugent.be> # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -29,6 +30,7 @@ import os import stat import time import shutil +from enum import Enum import rumba.log as log @@ -85,16 +87,24 @@ class Testbed(object): else: self.system_logs = system_logs - @abc.abstractmethod def swap_in(self, experiment): for node in experiment.nodes: node.executor = self.executor + self._swap_in(experiment) + + for dif in experiment.dif_ordering: + if isinstance(dif, ShimEthDIF): + dif.link_quality.apply(dif) + + @abc.abstractmethod + def _swap_in(self, experiment): + logger.info("_swap_in(): nothing to do") + @abc.abstractmethod def swap_out(self, experiment): logger.info("swap_out(): nothing to do") - # Base class for DIFs # # @name [string] DIF name @@ -149,11 +159,9 @@ class ShimEthDIF(DIF): def get_e_id(self): return "ShimEthDIF." + self.name - def __init__(self, name, members=None, link_speed=0): + def __init__(self, name, members=None, link_quality=None): DIF.__init__(self, name, members) - self.link_speed = int(link_speed) - if self.link_speed < 0: - raise ValueError("link_speed must be a non-negative number") + self._link_quality = link_quality def get_ipcp_class(self): return ShimEthIPCP @@ -163,6 +171,24 @@ class ShimEthDIF(DIF): if len(self.members) > 2: raise Exception("More than 2 members in %s!" % self.name) + @property + def link_quality(self): + return self._link_quality + + @link_quality.setter + def link_quality(self, _link_quality): + if not _link_quality: + raise ValueError("Cannot set link_quality to None, use del " + "link_quality to reset") + + self._link_quality = _link_quality + + _link_quality.apply(self) + + @link_quality.deleter + def link_quality(self): + self._link_quality.deactivate(self) + # Normal DIF # @@ -193,6 +219,180 @@ class NormalDIF(DIF): return s +class Distribution(Enum): + NORMAL = 1 + PARETO = 2 + PARETONORMAL = 3 + + +class Delay(object): + def __init__(self, delay=0, jitter=None, correlation=None, + distribution=None): + """ + Configure link delay + :param delay: average delay in ms + :param jitter: jitter in ms + :param correlation: correlation in % + :param distribution: delay distribution, defaults to a Normal + distribution + """ + + if delay < 0: + raise ValueError("Delay needs to be at least 0") + + if jitter and not jitter > 0: + raise ValueError("Jitter needs to be higher than 0") + + if (not jitter) and correlation: + raise ValueError("Correlation requires a value for jitter") + + if correlation and (correlation < 0 or correlation > 100): + raise ValueError("Correlation needs to be between 0 and 100") + + self._delay = delay + self._jitter = jitter + self._correlation = correlation + self._distribution = distribution + + @property + def delay(self): + return self._delay + + @property + def jitter(self): + return self._jitter + + @property + def correlation(self): + return self._correlation + + @property + def distribution(self): + return self._distribution + + def build_command(self): + opts = ["delay %ims" % self.delay] + + if self.jitter: + opts.append("%ims" % self.jitter) + + if self.correlation: + opts.append("%f%%" % self.correlation) + + if self.distribution: + opts.append("distribution %s" % self.distribution.name.lower()) + + return " ".join(opts) + + +class Loss(object): + def __init__(self, loss, correlation=None): + """ + Configure link loss + :param loss: loss in percentage + :param correlation: correlation in percentage + """ + if loss and (loss < 0 or loss > 100): + raise ValueError("Loss needs to be between 0 and 100") + + if correlation and (correlation < 0 or correlation > 100): + raise ValueError("Correlation needs to be between 0 and 100") + + self._loss = loss + self._correlation = correlation + + @property + def loss(self): + return self._loss + + @property + def correlation(self): + return self._correlation + + def build_command(self): + opts = ["loss %f%%" % self.loss] + + if self.correlation: + opts.append("%f%%" % self.correlation) + + return " ".join(opts) + + +class LinkQuality(object): + _active = set() + + def __init__(self, delay=None, loss=None, rate=None): + """ + Link quality configuration + :param delay: Delay object holding delay configuration + :param loss: Loss object holding delay configuration + :param rate: The rate of the link in mbit + """ + + if rate and not rate > 0: + raise ValueError("Rate needs to be higher than 0") + + self._delay = delay + self._loss = loss + self._rate = rate + + @property + def delay(self): + return self._delay + + @property + def loss(self): + return self._loss + + @property + def rate(self): + return self._rate + + def build_command(self, ipcp): + cmd = [] + + if ipcp in LinkQuality._active: + cmd.append("tc qdisc change dev %s root netem" % ipcp.ifname) + else: + cmd.append("tc qdisc add dev %s root netem" % ipcp.ifname) + + if self.delay: + cmd.append(self.delay.build_command()) + + if self.loss: + cmd.append(self.loss.build_command()) + + if self.rate: + cmd.append("rate %imbit" % self.rate) + + return " ".join(cmd) + + def apply(self, shim): + if not (self.delay or self.loss or self.rate): + self.deactivate(shim) + else: + for ipcp in shim.ipcps: + if not ipcp.ifname: + logger.error("Could not apply LinkQuality to IPCP because " + "the interface name is None") + continue + + ipcp.node.execute_command(self.build_command(ipcp), + as_root=True) + LinkQuality._active.add(ipcp) + + def deactivate(self, shim): + for ipcp in shim.ipcps: + if not ipcp.ifname: + logger.error("Could not remove LinkQuality from IPCP because " + "the interface name is None") + continue + + ipcp.node.execute_command("tc qdisc del dev %s root " + "netem" % ipcp.ifname, as_root=True) + LinkQuality._active.remove(ipcp) + + # SSH Configuration # class SSHConfig(object): diff --git a/rumba/testbeds/dockertb.py b/rumba/testbeds/dockertb.py index 3895a7e..d018f8c 100644 --- a/rumba/testbeds/dockertb.py +++ b/rumba/testbeds/dockertb.py @@ -33,7 +33,6 @@ import rumba.model as mod import rumba.log as log from rumba.executors.docker import DockerExecutor - logger = log.get_logger(__name__) class Testbed(mod.Testbed): @@ -56,11 +55,9 @@ class Testbed(mod.Testbed): self.docker_client = docker.from_env() self.executor = DockerExecutor(self) - def swap_in(self, experiment): + def _swap_in(self, experiment): docker_client = self.docker_client - mod.Testbed.swap_in(self, experiment) - # Pull image if self.pull_image: docker_client.images.pull(self.base_image_repo, diff --git a/rumba/testbeds/emulab.py b/rumba/testbeds/emulab.py index 06e0e5c..0d18892 100644 --- a/rumba/testbeds/emulab.py +++ b/rumba/testbeds/emulab.py @@ -271,16 +271,13 @@ class Testbed(mod.Testbed): if self.ip[ipcp] == item[1]: ipcp.ifname = item[0] - def swap_in(self, experiment): - mod.Testbed.swap_in(self, experiment) - + def _swap_in(self, experiment): self._create_experiment(experiment) wait = self.swap_exp_in() if wait: self.wait_until_nodes_up() self.complete_experiment_graph(experiment) - def swap_out(self, experiment): """ Swaps experiment out diff --git a/rumba/testbeds/jfed.py b/rumba/testbeds/jfed.py index db9dd14..a1ceded 100644 --- a/rumba/testbeds/jfed.py +++ b/rumba/testbeds/jfed.py @@ -206,9 +206,7 @@ class Testbed(mod.Testbed): logger.error("jFed returned with error " + str(e.returncode)) raise - def swap_in(self, experiment): - mod.Testbed.swap_in(self, experiment) - + def _swap_in(self, experiment): for node in experiment.nodes: node.ssh_config.set_http_proxy(self.http_proxy) self.create_rspec(experiment) diff --git a/rumba/testbeds/local.py b/rumba/testbeds/local.py index 77aed82..7c1aab1 100644 --- a/rumba/testbeds/local.py +++ b/rumba/testbeds/local.py @@ -40,9 +40,7 @@ class Testbed(mod.Testbed): self.executor = LocalExecutor(self) - def swap_in(self, experiment): - mod.Testbed.swap_in(self, experiment) - + def _swap_in(self, experiment): logger.info("Experiment swapped in") def swap_out(self, experiment): diff --git a/rumba/testbeds/qemu.py b/rumba/testbeds/qemu.py index df80b7e..5f76684 100644 --- a/rumba/testbeds/qemu.py +++ b/rumba/testbeds/qemu.py @@ -125,14 +125,12 @@ class Testbed(mod.Testbed): "ip link set dev %(ifname)s up" % {'ifname': ipcp.ifname}) - def swap_in(self, experiment): + def _swap_in(self, experiment): """ :type experiment mod.Experiment :param experiment: The experiment running """ - mod.Testbed.swap_in(self, experiment) - if os.geteuid() != 0: try: subprocess.check_call(["sudo", "-v"]) @@ -195,19 +193,6 @@ class Testbed(mod.Testbed): 'brctl addif %(br)s %(tap)s' % {'tap': tap_id, 'br': shim.name} ).split('\n') - if shim.link_speed > 0: - speed = '%dmbit' % shim.link_speed - - # Rate limit the traffic transmitted on the TAP interface - command_list += ( - 'tc qdisc add dev %(tap)s handle 1: root ' - 'htb default 11\n' - 'tc class add dev %(tap)s parent 1: classid ' - '1:1 htb rate 10gbit\n' - 'tc class add dev %(tap)s parent 1:1 classid ' - '1:11 htb rate %(speed)s' - % {'tap': tap_id, 'speed': speed} - ).split('\n') # While we're at it, build vm ports table and ipcp table vm['ports'].append({'tap_id': tap_id, |