updated how distributed servers are added and connections are created to reduce duplicate connections

daemon/core/emulator/session.py

@@ -147,7 +147,7 @@ class Session(object):
         self.sdt = Sdt(session=self)
 
         # distributed servers
-        self.servers = set()
+        self.servers = {}
 
         # initialize default node services
         self.services.default_services = {
@@ -158,10 +158,21 @@ class Session(object):
             "host": ("DefaultRoute", "SSH"),
         }
 
+    def add_distributed(self, server):
+        conn = Connection(server, user="root")
+        self.servers[server] = conn
+
     def init_distributed(self):
         for server in self.servers:
+            conn = self.servers[server]
             cmd = "mkdir -p %s" % self.session_dir
-            Connection(server, user="root").run(cmd, hide=False)
+            conn.run(cmd, hide=False)
+
+    def shutdown_distributed(self):
+        for server in self.servers:
+            conn = self.servers[server]
+            cmd = "rm -rf %s" % self.session_dir
+            conn.run(cmd, hide=False)
 
     @classmethod
     def get_node_class(cls, _type):
@@ -676,6 +687,13 @@ class Session(object):
         if not name:
             name = "%s%s" % (node_class.__name__, _id)
 
+        # verify distributed server
+        server = self.servers.get(node_options.emulation_server)
+        if node_options.emulation_server is not None and server is None:
+            raise CoreError(
+                "invalid distributed server: %s" % node_options.emulation_server
+            )
+
         # create node
         logging.info(
             "creating node(%s) id(%s) name(%s) start(%s)",
@@ -694,11 +712,7 @@ class Session(object):
             )
         else:
             node = self.create_node(
-                cls=node_class,
+                cls=node_class, _id=_id, name=name, start=start, server=server
-                _id=_id,
-                name=name,
-                start=start,
-                server=node_options.emulation_server,
             )
 
         # set node attributes
@@ -972,6 +986,7 @@ class Session(object):
         preserve = self.options.get_config("preservedir") == "1"
         if not preserve:
             shutil.rmtree(self.session_dir, ignore_errors=True)
+            self.shutdown_distributed()
 
         # call session shutdown handlers
         for handler in self.shutdown_handlers:

daemon/core/nodes/base.py

@@ -13,8 +13,6 @@ from builtins import range
 from socket import AF_INET, AF_INET6
 from tempfile import NamedTemporaryFile
 
-from fabric import Connection
-
 from core import constants, utils
 from core.emulator.data import LinkData, NodeData
 from core.emulator.enumerations import LinkTypes, NodeTypes
@@ -42,7 +40,8 @@ class NodeBase(object):
         :param int _id: id
         :param str name: object name
         :param bool start: start value
-        :param str server: remote server node will run on, default is None for localhost
+        :param fabric.connection.Connection server: remote server node will run on,
+            default is None for localhost
         """
 
         self.session = session
@@ -53,8 +52,6 @@ class NodeBase(object):
             name = "o%s" % self.id
         self.name = name
         self.server = server
-        if self.server is not None:
-            self.server_conn = Connection(self.server, user="root")
 
         self.type = None
         self.services = None
@@ -103,18 +100,23 @@ class NodeBase(object):
             return utils.check_cmd(args, env=env)
         else:
             args = " ".join(args)
-            return self.remote_cmd(args)
+            return self.remote_cmd(args, env=env)
 
-    def remote_cmd(self, cmd):
+    def remote_cmd(self, cmd, env=None):
         """
         Run command remotely using server connection.
 
         :param str cmd: command to run
+        :param dict env: environment for remote command, default is None
         :return: stdout when success
         :rtype: str
         :raises CoreCommandError: when a non-zero exit status occurs
         """
-        result = self.server_conn.run(cmd, hide=False)
+        if env is None:
+            result = self.server.run(cmd, hide=False)
+        else:
+            logging.info("command env: %s", env)
+            result = self.server.run(cmd, hide=False, env=env, replace_env=True)
         if result.exited:
             raise CoreCommandError(
                 result.exited, result.command, result.stdout, result.stderr
@@ -969,7 +971,7 @@ class CoreNode(CoreNodeBase):
             self.client.check_cmd(["sync"])
         else:
             self.net_cmd(["mkdir", "-p", directory])
-            self.server_conn.put(srcname, filename)
+            self.server.put(srcname, filename)
 
     def hostfilename(self, filename):
         """
@@ -992,7 +994,7 @@ class CoreNode(CoreNodeBase):
         Create a node file with a given mode.
 
         :param str filename: name of file to create
-        :param contents: contents of file
+        :param str contents: contents of file
         :param int mode: mode for file
         :return: nothing
         """
@@ -1005,12 +1007,12 @@ class CoreNode(CoreNodeBase):
                 open_file.write(contents)
                 os.chmod(open_file.name, mode)
         else:
-            temp = NamedTemporaryFile()
+            temp = NamedTemporaryFile(delete=False)
-            temp.write(contents)
+            temp.write(contents.encode("utf-8"))
             temp.close()
-            self.net_cmd(["mkdir", "-m", oct(0o755), "-p", dirname])
+            self.net_cmd(["mkdir", "-m", "%o" % 0o755, "-p", dirname])
-            self.server_conn.put(temp.name, hostfilename)
+            self.server.put(temp.name, hostfilename)
-            self.net_cmd(["chmod", oct(mode), hostfilename])
+            self.net_cmd(["chmod", "%o" % mode, hostfilename])
         logging.debug(
             "node(%s) added file: %s; mode: 0%o", self.name, hostfilename, mode
         )
@@ -1031,9 +1033,9 @@ class CoreNode(CoreNodeBase):
             if mode is not None:
                 os.chmod(hostfilename, mode)
         else:
-            self.server_conn.put(srcfilename, hostfilename)
+            self.server.put(srcfilename, hostfilename)
             if mode is not None:
-                self.net_cmd(["chmod", oct(mode), hostfilename])
+                self.net_cmd(["chmod", "%o" % mode, hostfilename])
         logging.info(
             "node(%s) copied file: %s; mode: %s", self.name, hostfilename, mode
         )

daemon/examples/python/distributed.py

@@ -1,4 +1,5 @@
 import logging
+import pdb
 
 from core.emulator.coreemu import CoreEmu
 from core.emulator.emudata import NodeOptions
@@ -14,7 +15,7 @@ def main():
     session = coreemu.create_session()
 
     # initialize distributed
-    session.servers.add("core2")
+    session.add_distributed("core2")
     session.init_distributed()
 
     # must be in configuration state for nodes to start, when using "node_add" below
@@ -25,13 +26,12 @@ def main():
 
     # create nodes
     options = NodeOptions()
-    options.emulation_server = "10.10.4.38"
     options.emulation_server = "core2"
     session.add_node(node_options=options)
     # interface = prefixes.create_interface(node_one)
     # session.add_link(node_one.id, switch.id, interface_one=interface)
 
-    # node_two = session.add_node()
+    session.add_node()
     # interface = prefixes.create_interface(node_two)
     # session.add_link(node_two.id, switch.id, interface_one=interface)
 
@@ -46,6 +46,8 @@ def main():
     # node_two.client.icmd(["iperf", "-t", "10", "-c", node_one_address])
     # node_one.cmd(["killall", "-9", "iperf"])
 
+    pdb.set_trace()
+
     # shutdown session
     coreemu.shutdown()