prototypes: Add local Ouroboros support

This adds local Ouroboros support in case of a fake testbed (i.e. no
resources allocated). A next PR will abstract away the mode of
communication with the testbed, since the Docker plugin will also need
it.

This also adds another function to the model, terminate_prototype,
which should clean up the prototype gracefully, or can be skipped
depending on the testbed.

Currently the ouroboros plugin with the fake testbed needs to be run
as root. If there is a way to run the command as root in the
background and then clean it up properly, I would be happy to know
how.

5 files changed, 75 insertions, 21 deletions

diff --git a/rumba/model.py b/rumba/model.py
index 2749ae7..42112be 100644
--- a/rumba/model.py
+++ b/rumba/model.py
@@ -874,6 +874,10 @@ class Experiment(object):
     def prototype_name(self):
         raise Exception('prototype_name() method not implemented')
 
+    @abc.abstractmethod
+    def _terminate_prototype(self):
+        raise Exception('terminate_prototype() method not implemented')
+
     def swap_in(self):
         # Realize the experiment testbed (testbed-specific)
         start = time.time()
@@ -884,6 +888,8 @@ class Experiment(object):
 
     def swap_out(self):
         start = time.time()
+        # Terminate prototype gracefully
+        self._terminate_prototype()
         for node in self.nodes:
             if node.ssh_config.client is not None:
                 node.ssh_config.client.close()
diff --git a/rumba/prototypes/irati.py b/rumba/prototypes/irati.py
index bf3ce8c..0807e10 100644
--- a/rumba/prototypes/irati.py
+++ b/rumba/prototypes/irati.py
@@ -426,3 +426,6 @@ class Experiment(mod.Experiment):
                     conf_files.setdefault(node, []).append(
                         'normal.%s.%s.dif' % (node.name, dif.name))
         return conf_files
+
+    def _terminate_prototype(self):
+        return
diff --git a/rumba/prototypes/ouroboros.py b/rumba/prototypes/ouroboros.py
index 6dec9e4..c588405 100644
--- a/rumba/prototypes/ouroboros.py
+++ b/rumba/prototypes/ouroboros.py
@@ -25,11 +25,13 @@
 #
 
 import time
+import subprocess
 
 import rumba.ssh_support as ssh
 import rumba.model as mod
 import rumba.multiprocess as m_processing
 import rumba.log as log
+import rumba.testbeds.faketestbed as fake
 
 
 logger = log.get_logger(__name__)
@@ -53,13 +55,33 @@ class Experiment(mod.Experiment):
     def prototype_name(self):
         return 'ouroboros'
 
+    def exec_local_cmd(self, cmd):
+        try:
+            logger.info(cmd)
+            subprocess.check_call(cmd.split(' '))
+        except subprocess.CalledProcessError as e:
+            logger.error("Return code was " + str(e.returncode))
+            raise
+
+    def exec_local_cmds(self, cmds):
+        for cmd in cmds:
+            self.exec_local_cmd(cmd)
+
     def setup_ouroboros(self):
-        for node in self.nodes:
-            ssh.execute_command(self.testbed, node.ssh_config,
-                                "sudo nohup irmd > /dev/null &",
-                                time_out=None)
+        if isinstance(self.testbed, fake.Testbed):
+            self.irmd = subprocess.Popen(["irmd"])
+            logger.info("Started IRMd, sleeping 2 seconds...")
+            time.sleep(2)
+        else:
+            for node in self.nodes:
+                ssh.execute_command(self.testbed, node.ssh_config,
+                                    "sudo nohup irmd > /dev/null &",
+                                    time_out=None)
 
     def install_ouroboros(self):
+        if isinstance(self.testbed, fake.Testbed):
+            return
+
         packages = ["cmake", "protobuf-c-compiler", "git", "libfuse-dev",
                     "libgcrypt20-dev", "libssl-dev"]
 
@@ -97,11 +119,11 @@ class Experiment(mod.Experiment):
                     cmd = "irm i c n " + ipcp.name
 
                 if isinstance(ipcp.dif, mod.ShimEthDIF):
-                    # NOTE: Here to test with fake testbed
-                    if ipcp.ifname is None:
-                        ipcp.ifname = "eth0"
-                    cmd += " type eth-llc if_name " + ipcp.ifname
-                    cmd += " layer " + ipcp.dif.name
+                    if isinstance(self.testbed, fake.Testbed):
+                        cmd += " type local layer " + ipcp.dif.name
+                    else:
+                        cmd += " type eth-llc if_name " + ipcp.ifname
+                        cmd += " layer " + ipcp.dif.name
                 elif isinstance(ipcp.dif, mod.NormalDIF):
                     cmd += " type normal"
                     if ipcp.dif_bootstrapper:
@@ -131,8 +153,11 @@ class Experiment(mod.Experiment):
                 # Postpone registrations
                 self.r_ipcps[ipcp] = cmds2
 
-            ssh.execute_commands(self.testbed, node.ssh_config, cmds,
-                                 time_out=None)
+            if isinstance(self.testbed, fake.Testbed):
+                self.exec_local_cmds(cmds)
+            else:
+                ssh.execute_commands(self.testbed, node.ssh_config, cmds,
+                                     time_out=None)
 
     def enroll_dif(self, el):
         for e in el:
@@ -141,9 +166,13 @@ class Experiment(mod.Experiment):
 
             # Execute postponed registration
             if e['enroller'] in self.r_ipcps:
-                ssh.execute_commands(self.testbed,
-                                     e['enroller'].node.ssh_config,
-                                     self.r_ipcps[e['enroller']], time_out=None)
+                if isinstance(self.testbed, fake.Testbed):
+                    self.exec_local_cmds(self.r_ipcps[e['enroller']])
+                else:
+                    ssh.execute_commands(self.testbed,
+                                         e['enroller'].node.ssh_config,
+                                         self.r_ipcps[e['enroller']],
+                                         time_out=None)
                 self.r_ipcps.pop(e['enroller'], None)
 
             cmd = "irm r n " + ipcp.name
@@ -158,9 +187,12 @@ class Experiment(mod.Experiment):
                 cmd += " layer " + dif_b.name
             cmds.append(cmd)
 
-            ssh.execute_commands(self.testbed,
-                                 e['enrollee'].node.ssh_config,
-                                 cmds, time_out=None)
+            if isinstance(self.testbed, fake.Testbed):
+                self.exec_local_cmds(cmds)
+            else:
+                ssh.execute_commands(self.testbed,
+                                     e['enrollee'].node.ssh_config,
+                                     cmds, time_out=None)
 
     def setup_flows(self, el, comp):
         for e in el:
@@ -168,9 +200,12 @@ class Experiment(mod.Experiment):
             cmd = "irm i conn n " + ipcp.name + " comp " + \
                   comp + " dst " + e['dst'].name
 
-            ssh.execute_command(self.testbed,
-                                ipcp.node.ssh_config,
-                                cmd, time_out=None)
+            if isinstance(self.testbed, fake.Testbed):
+                self.exec_local_cmd(cmd)
+            else:
+                ssh.execute_command(self.testbed,
+                                    ipcp.node.ssh_config,
+                                    cmd, time_out=None)
 
     def _install_prototype(self):
         logger.info("Installing Ouroboros...")
@@ -192,3 +227,7 @@ class Experiment(mod.Experiment):
             self.setup_flows(dt, comp="dt")
 
         logger.info("All done, have fun!")
+
+    def _terminate_prototype(self):
+        if isinstance(self.testbed, fake.Testbed):
+            self.irmd.terminate()
diff --git a/rumba/prototypes/rlite.py b/rumba/prototypes/rlite.py
index dca693b..e1b3745 100644
--- a/rumba/prototypes/rlite.py
+++ b/rumba/prototypes/rlite.py
@@ -176,3 +176,6 @@ class Experiment(mod.Experiment):
         logger.info("IPCPs registered to their lower DIFs on all nodes")
         self.enroll_ipcps()
         logger.info("enrollment completed in all DIFs")
+
+    def _terminate_prototype(self):
+        return
diff --git a/rumba/testbeds/faketestbed.py b/rumba/testbeds/faketestbed.py
index 44994e3..b021aca 100644
--- a/rumba/testbeds/faketestbed.py
+++ b/rumba/testbeds/faketestbed.py
@@ -38,4 +38,7 @@ class Testbed(mod.Testbed):
         mod.Testbed.__init__(self, exp_name, username, password, proj_name)
 
     def swap_in(self, experiment):
-        logger.info("[Fake testbed] experiment swapped in")
+        logger.info("Experiment swapped in")
+
+    def swap_out(self, experiment):
+        logger.info("Experiment swapped out")