diff options
| -rw-r--r-- | rumba/elements/experimentation.py | 16 | ||||
| -rw-r--r-- | rumba/elements/topology.py | 98 | ||||
| -rw-r--r-- | rumba/irm_backend.py | 8 | ||||
| -rw-r--r-- | rumba/prototypes/ouroboros.py | 35 | ||||
| -rw-r--r-- | rumba/testbeds/local.py | 4 | ||||
| -rw-r--r-- | rumba/testbeds/localnet.py | 5 | ||||
| -rw-r--r-- | rumba/utils.py | 14 |
7 files changed, 134 insertions, 46 deletions
diff --git a/rumba/elements/experimentation.py b/rumba/elements/experimentation.py index 67fb2e3..0b53f11 100644 --- a/rumba/elements/experimentation.py +++ b/rumba/elements/experimentation.py @@ -84,6 +84,22 @@ class Testbed(object): # testbed) have no interfaces, so applying a LinkQuality is a no-op. supports_link_quality = True + # Whether rumba must build/install the prototype on the nodes. Testbeds + # that run on the host against already-installed binaries (local, localnet, + # nstb) set this False; remote testbeds (emulab, jfed, ...) leave it True. + needs_install = True + + # Whether the prototype runs as a single instance in one namespace/host: + # one IRMd, driven as local subprocesses / the in-process IRM backend. + # Testbeds that give each node its own namespace (nstb) or its own + # container/host (docker, remote) set this False. + single_irmd = False + + # Whether swap_out destroys the execution context (e.g. network + mount + # namespaces) the prototype runs in, so the prototype must be terminated + # *before* swap_out while it is still reachable. nstb sets this True. + terminate_before_swapout = False + def __init__(self, exp_name, username, diff --git a/rumba/elements/topology.py b/rumba/elements/topology.py index f6ad32a..f69ae29 100644 --- a/rumba/elements/topology.py +++ b/rumba/elements/topology.py @@ -745,44 +745,76 @@ class LinkQuality(object): def corrupt(self): return self._corrupt + @staticmethod + def _netem_target(ipcp): + """Return (netem_dev, on_host). + + netem goes on the bridge-side interface (host_ifname) when there is + one, else on the IPCP interface itself. *on_host* is True when that + interface lives on the host rather than in the IPCP's namespace, in + which case the command must run through the node's host_executor. + """ + host_ifname = getattr(ipcp, 'host_ifname', None) + if host_ifname and host_ifname != ipcp.ifname: + return host_ifname, True + return ipcp.ifname, False + + @staticmethod + def _run_host(ipcp, cmds, check=True): + """Run host-side (bridge/netem) commands. + + Uses the node's host_executor when set (namespaced testbeds, where + the bridge-side interface is on the host); otherwise falls back to + the node executor, which for the non-namespaced testbeds already runs + on the host -- so behaviour is unchanged there. + """ + host_executor = getattr(ipcp.node, 'host_executor', None) + for cmd in cmds: + if host_executor is not None: + host_executor.execute_command(ipcp.node, cmd, True, + check=check) + else: + ipcp.node.execute_command(cmd, as_root=True, check=check) + def build_commands(self, ipcp): - netem_cmd = [] - cmds = [] - - # Rate limiting (HTB) goes on the IPCP interface so a blocking - # send back-pressures the source. All netem impairments go on - # the bridge-side interface (host_ifname), where frames are - # already orphaned -- netem's skb_orphan() there no longer - # defeats that back-pressure. Testbeds without a bridge (e.g. - # UDP veths) have no host_ifname and put both on the IPCP - # interface. apply_to_ipcp() clears any prior qdiscs first, so - # these are plain adds. - netem_dev = getattr(ipcp, 'host_ifname', None) or ipcp.ifname + """Split tc commands into (node_cmds, host_cmds). + + Rate limiting (HTB) goes on the IPCP interface so a blocking send + back-pressures the source, and always runs in the IPCP's namespace. + netem impairments go on the bridge-side interface (host_ifname) where + frames are already orphaned -- netem's skb_orphan() there no longer + defeats that back-pressure -- and run on the host for namespaced + testbeds. Testbeds without a bridge (e.g. UDP veths) have no + host_ifname and chain netem under HTB on the IPCP interface. + apply_to_ipcp() clears any prior qdiscs first, so these are plain adds. + """ + node_cmds = [] + host_cmds = [] + netem_dev, on_host = self._netem_target(ipcp) if self.rate: - cmds.append( + node_cmds.append( "tc qdisc add dev %s root handle 1: htb default 1" % ipcp.ifname) - cmds.append( + node_cmds.append( "tc class add dev %s parent 1: classid 1:1 " "htb rate %imbit" % (ipcp.ifname, self.rate)) netem_parts = [self.delay, self.loss, self.duplicate, self.reorder, self.corrupt] if any(netem_parts): - if self.rate and netem_dev == ipcp.ifname: + if self.rate and not on_host: qref = "parent 1:1" # no bridge: chain under HTB else: qref = "root" - netem_cmd.append( - "tc qdisc add dev %s %s netem limit 65536" - % (netem_dev, qref)) + netem_cmd = ["tc qdisc add dev %s %s netem limit 65536" + % (netem_dev, qref)] for part in netem_parts: if part: netem_cmd.append(part.build_command()) - cmds.append(" ".join(netem_cmd)) + (host_cmds if on_host else node_cmds).append(" ".join(netem_cmd)) - return cmds + return node_cmds, host_cmds def apply_to_ipcp(self, ipcp): """Apply this link quality configuration to a single *ipcp*.""" @@ -803,8 +835,11 @@ class LinkQuality(object): # Clear any prior qdiscs first (best-effort, both devices) so # the adds in build_commands never hit an existing root qdisc. self.deactivate_ipcp(ipcp) - ipcp.node.execute_commands( - self.build_commands(ipcp), as_root=True) + node_cmds, host_cmds = self.build_commands(ipcp) + if node_cmds: + ipcp.node.execute_commands(node_cmds, as_root=True) + if host_cmds: + self._run_host(ipcp, host_cmds) ipcp._netem_active = True def deactivate_ipcp(self, ipcp): @@ -825,16 +860,16 @@ class LinkQuality(object): "Could not remove LinkQuality from IPCP because " "the interface name is None") return - # Best-effort teardown: HTB lives on the IPCP interface, netem on - # the bridge-side interface; either may be absent, so do not check. + # Best-effort teardown: HTB lives on the IPCP interface (node ns), + # netem on the bridge-side interface (host for namespaced testbeds); + # either may be absent, so do not check. ipcp.node.execute_command( "tc qdisc del dev %s root" % ipcp.ifname, as_root=True, check=False) - host_ifname = getattr(ipcp, 'host_ifname', None) - if host_ifname and host_ifname != ipcp.ifname: - ipcp.node.execute_command( - "tc qdisc del dev %s root" % host_ifname, - as_root=True, check=False) + netem_dev, on_host = self._netem_target(ipcp) + if on_host: + self._run_host( + ipcp, ["tc qdisc del dev %s root" % netem_dev], check=False) def apply(self, layer): """Apply this link quality to all IPCPs in *layer*.""" @@ -916,6 +951,11 @@ class Node(object): self.policies[layer] = policies.get(layer, Policy(layer, self)) self.executor = None # set by testbed on swap_in + # Optional executor for host-side link shaping. When a testbed runs + # nodes in their own namespace (nstb), the bridge-side (host_ifname) + # interface lives on the host, not in the node's namespace, so netem + # must be applied there. None means "same namespace as executor". + self.host_executor = None self.startup_command = None # set by prototype self._validate() diff --git a/rumba/irm_backend.py b/rumba/irm_backend.py index 2550884..07a51a7 100644 --- a/rumba/irm_backend.py +++ b/rumba/irm_backend.py @@ -184,13 +184,21 @@ class IrmCLI(IrmBackend): node.execute_command(cmd, time_out=None) + def _ensure_name(self, node, name): + # Allocating a flow to a name requires that name to exist in the + # local IRMd registry; with one IRMd per node a remote destination + # is unknown until created. No-op when it already exists. + node.execute_command("irm n c %s || true" % name, time_out=None) + def enroll_ipcp(self, node, name, dst_name, autobind=True): + self._ensure_name(node, dst_name) cmd = "irm i e n %s dst %s" % (name, dst_name) if autobind: cmd += " autobind" node.execute_command(cmd, time_out=None) def connect_ipcp(self, node, name, dst_name): + self._ensure_name(node, dst_name) cmd = "irm i conn n %s dst %s" % (name, dst_name) node.execute_command(cmd, time_out=None) diff --git a/rumba/prototypes/ouroboros.py b/rumba/prototypes/ouroboros.py index 1c2e007..931b2e0 100644 --- a/rumba/prototypes/ouroboros.py +++ b/rumba/prototypes/ouroboros.py @@ -146,9 +146,10 @@ class Experiment(mod.Experiment): self.set_startup_command("irmd") self.metrics_python_version = "python3.9" - # Create IRM backend - if isinstance(testbed, (local.Testbed, - localnet.LocalNetworkTestbed)): + # Create IRM backend. IrmPython drives a single in-process IRMd, so + # it only makes sense for single-IRMd testbeds; everything else (per + # node/namespace/host IRMds) is driven through the CLI backend. + if testbed.single_irmd: try: self.irm = IrmPython() logger.info("Using pyouroboros IRM backend") @@ -192,7 +193,7 @@ class Experiment(mod.Experiment): except ImportError: pass - if self._is_local: + if self.testbed.single_irmd: subprocess.check_call('sudo -v'.split()) popen = subprocess.Popen(["sudo", "irmd"]) self.irmd = Process(None, popen.pid, "irmd", popen=popen) @@ -201,8 +202,16 @@ class Experiment(mod.Experiment): time.sleep(2) else: for node in self.nodes: - node.execute_command("sudo nohup irmd > /dev/null &", - time_out=None) + # Redirect BOTH stdout and stderr: irmd does not daemonize, so + # a leaked stderr fd would keep an output-capturing executor + # (e.g. the nsenter executor) blocked forever waiting for EOF. + # --stdout: without it irmd logs to syslog and the file + # stays empty; with it every message is fflush()ed, so the + # log (in the node's private tmpfs) can be read live. + node.execute_command( + "sudo nohup irmd --stdout " + "> /var/run/ouroboros/irmd.log 2>&1 &", + time_out=None) time.sleep(1) def _install_packages_and_execute_cmds(self, packages, cmds, nodes=None): @@ -226,14 +235,8 @@ class Experiment(mod.Experiment): m_processing.call_in_parallel(names, args, executors) - @property - def _is_local(self): - """True if the testbed runs locally (single IRMd).""" - return isinstance(self.testbed, - (local.Testbed, localnet.LocalNetworkTestbed)) - def install_ouroboros(self): - if self._is_local: + if not self.testbed.needs_install: return packages = ["cmake", "protobuf-c-compiler", "git", "libfuse-dev", @@ -250,7 +253,7 @@ class Experiment(mod.Experiment): self._install_packages_and_execute_cmds(packages, cmds) def update_ouroboros(self, branch): - if self._is_local: + if not self.testbed.needs_install: return if branch is not None: @@ -284,7 +287,7 @@ class Experiment(mod.Experiment): self._install_packages_and_execute_cmds(packages, []) def install_ouroboros_python_exporter(self): - if self._is_local: + if not self.testbed.needs_install: return if self.influxdb is None: @@ -510,7 +513,7 @@ class Experiment(mod.Experiment): cmds.append('killall -15 irmd') logger.info("Killing Ouroboros...") - if self._is_local: + if self.testbed.single_irmd: cmds = list(map(lambda c: "sudo %s" % (c,), cmds)) for cmd in cmds: subprocess.call(cmd.split()) diff --git a/rumba/testbeds/local.py b/rumba/testbeds/local.py index ad0db9f..968c365 100644 --- a/rumba/testbeds/local.py +++ b/rumba/testbeds/local.py @@ -43,6 +43,10 @@ class Testbed(mod.Testbed): # (rate/delay/loss) cannot be applied -- treat apply as a no-op. supports_link_quality = False + # Runs against host-installed binaries as a single local IRMd. + needs_install = False + single_irmd = True + def __init__(self, exp_name='foo', username='bar', proj_name="rumba", password=""): """ diff --git a/rumba/testbeds/localnet.py b/rumba/testbeds/localnet.py index 4278367..28ca080 100644 --- a/rumba/testbeds/localnet.py +++ b/rumba/testbeds/localnet.py @@ -52,6 +52,11 @@ class LocalNetworkTestbed(mod.Testbed): netem support on the veth interfaces. """ + # Runs against host-installed binaries as a single local IRMd, in the + # default namespace (all veths/bridges live on the host). + needs_install = False + single_irmd = True + def __init__(self, exp_name='localnet', username='', proj_name='rumba', password=''): """ diff --git a/rumba/utils.py b/rumba/utils.py index 324d3bf..6742f4d 100644 --- a/rumba/utils.py +++ b/rumba/utils.py @@ -211,10 +211,22 @@ class ExperimentManager(object): # Swap out if do_swap_out: + testbed = self.experiment.testbed # Kill prototype in case of local testbed - if isinstance(self.experiment.testbed, local.Testbed): + if isinstance(testbed, local.Testbed): self.experiment.terminate_prototype() else: + # Testbeds whose swap_out destroys the execution context + # (e.g. nstb's per-node net+mount namespaces) must reap the + # prototype first, while it is still reachable. + if testbed.terminate_before_swapout: + try: + self.experiment.terminate_prototype() + except Exception as e: + logger.warning( + 'Could not terminate prototype before swap ' + 'out. %s%s', type(e).__name__, + ": " + str(e) if str(e) != "" else "") self.experiment.swap_out() if exc_val is not None: logger.error('Something went wrong. ' |