from nepi.execution.attribute import Attribute, Flags, Types
from nepi.execution.resource import clsinit_copy, ResourceState, \
reschedule_delay
-from nepi.resources.linux.application import LinuxApplication
+from nepi.resources.linux.tunnel import LinuxTunnel
from nepi.util.sshfuncs import ProcStatus
from nepi.util.timefuncs import tnow, tdiffsec
import time
@clsinit_copy
-class LinuxGRETunnel(LinuxApplication):
+class LinuxGRETunnel(LinuxTunnel):
_rtype = "LinuxGRETunnel"
_help = "Constructs a tunnel between two Linux endpoints using a UDP connection "
_backend = "linux"
- @classmethod
- def _register_attributes(cls):
- bwlimit = Attribute("bwLimit",
- "Specifies the interface's emulated bandwidth in bytes "
- "per second.",
- type = Types.Integer,
- flags = Flags.Design)
-
- cls._register_attribute(bwlimit)
-
- def __init__(self, ec, guid):
- super(LinuxGRETunnel, self).__init__(ec, guid)
-
def log_message(self, msg):
return " guid %d - GRE tunnel %s - %s - %s " % (self.guid,
self.endpoint1.node.get("hostname"),
connected = []
for guid in self.connections:
rm = self.ec.get_resource(guid)
- if hasattr(rm, "udp_connect_command"):
+ if hasattr(rm, "gre_connect_command"):
connected.append(rm)
return connected
- @property
- def endpoint1(self):
- endpoints = self.get_endpoints()
- if endpoints: return endpoints[0]
- return None
-
- @property
- def endpoint2(self):
- endpoints = self.get_endpoints()
- if endpoints and len(endpoints) > 1: return endpoints[1]
- return None
-
- def app_home(self, endpoint):
- return os.path.join(endpoint.node.exp_home, self._home)
-
- def run_home(self, endpoint):
- return os.path.join(self.app_home(endpoint), self.ec.run_id)
-
- def udp_connect(self, endpoint, remote_ip):
- # Get udp connect command
- local_port_file = os.path.join(self.run_home(endpoint),
- "local_port")
- remote_port_file = os.path.join(self.run_home(endpoint),
- "remote_port")
- ret_file = os.path.join(self.run_home(endpoint),
- "ret_file")
- cipher = self.get("cipher")
- cipher_key = self.get("cipherKey")
- bwlimit = self.get("bwLimit")
- txqueuelen = self.get("txQueueLen")
- udp_connect_command = endpoint.udp_connect_command(
- remote_ip, local_port_file, remote_port_file,
- ret_file, cipher, cipher_key, bwlimit, txqueuelen)
+ def initiate_connection(self, endpoint, remote_endpoint):
+ # Return the command to execute to initiate the connection to the
+ # other endpoint
+ connection_run_home = self.run_home(endpoint)
+ gre_connect_command = endpoint.gre_connect_command(
+ remote_endpoint, connection_run_home)
# upload command to connect.sh script
- shfile = os.path.join(self.app_home(endpoint), "udp-connect.sh")
+ shfile = os.path.join(self.app_home(endpoint), "gre-connect.sh")
endpoint.node.upload(udp_connect_command,
shfile,
text = True,
self.error(msg, out, err)
raise RuntimeError, msg
- # wait until port is written to file
- port = self.wait_local_port(endpoint)
- return (port, pid, ppid)
-
- def do_provision(self):
- # create run dir for tunnel on each node
- self.endpoint1.node.mkdir(self.run_home(self.endpoint1))
- self.endpoint2.node.mkdir(self.run_home(self.endpoint2))
-
- # Invoke connect script in endpoint 1
- remote_ip1 = socket.gethostbyname(self.endpoint2.node.get("hostname"))
- (port1, self._pid1, self._ppid1) = self.udp_connect(self.endpoint1,
- remote_ip1)
-
- # Invoke connect script in endpoint 2
- remote_ip2 = socket.gethostbyname(self.endpoint1.node.get("hostname"))
- (port2, self._pid2, self._ppid2) = self.udp_connect(self.endpoint2,
- remote_ip2)
-
- # upload file with port 2 to endpoint 1
- self.upload_remote_port(self.endpoint1, port2)
-
- # upload file with port 1 to endpoint 2
- self.upload_remote_port(self.endpoint2, port1)
-
- # check if connection was successful on both sides
- self.wait_result(self.endpoint1)
- self.wait_result(self.endpoint2)
-
- self.info("Provisioning finished")
-
- self.set_provisioned()
-
- def do_deploy(self):
- if (not self.endpoint1 or self.endpoint1.state < ResourceState.READY) or \
- (not self.endpoint2 or self.endpoint2.state < ResourceState.READY):
- self.ec.schedule(reschedule_delay, self.deploy)
- else:
- self.do_discover()
- self.do_provision()
-
- self.set_ready()
-
- def do_start(self):
- if self.state == ResourceState.READY:
- command = self.get("command")
- self.info("Starting command '%s'" % command)
-
- self.set_started()
- else:
- msg = " Failed to execute command '%s'" % command
- self.error(msg, out, err)
- raise RuntimeError, msg
-
- def do_stop(self):
- """ Stops application execution
- """
- if self.state == ResourceState.STARTED:
- self.info("Stopping tunnel")
-
- # Only try to kill the process if the pid and ppid
- # were retrieved
- if self._pid1 and self._ppid1 and self._pid2 and self._ppid2:
- (out1, err1), proc1 = self.endpoint1.node.kill(self._pid1,
- self._ppid1, sudo = True)
- (out2, err2), proc2 = self.endpoint2.node.kill(self._pid2,
- self._ppid2, sudo = True)
-
- if (proc1.poll() and err1) or (proc2.poll() and err2):
- # check if execution errors occurred
- msg = " Failed to STOP tunnel"
- self.error(msg, err1, err2)
- raise RuntimeError, msg
-
- self.set_stopped()
-
- @property
- def state(self):
- """ Returns the state of the application
- """
- if self._state == ResourceState.STARTED:
- # In order to avoid overwhelming the remote host and
- # the local processor with too many ssh queries, the state is only
- # requested every 'state_check_delay' seconds.
- state_check_delay = 0.5
- if tdiffsec(tnow(), self._last_state_check) > state_check_delay:
- if self._pid1 and self._ppid1 and self._pid2 and self._ppid2:
- # Make sure the process is still running in background
- # No execution errors occurred. Make sure the background
- # process with the recorded pid is still running.
- status1 = self.endpoint1.node.status(self._pid1, self._ppid1)
- status2 = self.endpoint2.node.status(self._pid2, self._ppid2)
-
- if status1 == ProcStatus.FINISHED and \
- status2 == ProcStatus.FINISHED:
-
- # check if execution errors occurred
- (out1, err1), proc1 = self.endpoint1.node.check_errors(
- self.run_home(self.endpoint1))
-
- (out2, err2), proc2 = self.endpoint2.node.check_errors(
- self.run_home(self.endpoint2))
-
- if err1 or err2:
- msg = "Error occurred in tunnel"
- self.error(msg, err1, err2)
- self.fail()
- else:
- self.set_stopped()
-
- self._last_state_check = tnow()
-
- return self._state
-
- def wait_local_port(self, endpoint):
- """ Waits until the local_port file for the endpoint is generated,
- and returns the port number
-
- """
- return self.wait_file(endpoint, "local_port")
-
- def wait_result(self, endpoint):
- """ Waits until the return code file for the endpoint is generated
-
- """
- return self.wait_file(endpoint, "ret_file")
-
- def wait_file(self, endpoint, filename):
- """ Waits until file on endpoint is generated """
- result = None
- delay = 1.0
+ # Wait if name
+ return True
- for i in xrange(20):
- (out, err), proc = endpoint.node.check_output(
- self.run_home(endpoint), filename)
+ def establish_connection(self, endpoint, remote_endpoint, data):
+ pass
- if out:
- result = out.strip()
- break
- else:
- time.sleep(delay)
- delay = delay * 1.5
- else:
- msg = "Couldn't retrieve %s" % filename
- self.error(msg, out, err)
- raise RuntimeError, msg
+ def verify_connection(self, endpoint, remote_endpoint):
+ # Execute a ping from both sides to verify that the tunnel works
+ pass
- return result
+ def terminate_connection(self, endpoint, remote_endpoint):
+ pass
- def upload_remote_port(self, endpoint, port):
- # upload remote port number to file
- port = "%s\n" % port
- endpoint.node.upload(port,
- os.path.join(self.run_home(endpoint), "remote_port"),
- text = True,
- overwrite = False)
+ def check_state_connection(self, endpoint, remote_endpoint):
+ raise NotImplementedError
def valid_connection(self, guid):
# TODO: Validate!
def get_endpoints(self):
""" Returns the list of RM that are endpoints to the tunnel
"""
- connected = []
- for guid in self.connections:
- rm = self.ec.get_resource(guid)
- if hasattr(rm, "udp_connect_command"):
- connected.append(rm)
- return connected
+ raise NotImplementedError
@property
def endpoint1(self):
self.endpoint2.node.get("hostname"),
msg)
+ def get_endpoints(self):
+ """ Returns the list of RM that are endpoints to the tunnel
+ """
+ connected = []
+ for guid in self.connections:
+ rm = self.ec.get_resource(guid)
+ if hasattr(rm, "udp_connect_command"):
+ connected.append(rm)
+ return connected
+
def initiate_connection(self, endpoint, remote_endpoint):
cipher = self.get("cipher")
cipher_key = self.get("cipherKey")
(out, err), proc = endpoint.node.run(cmd, self.run_home(endpoint))
# check if execution errors occurred
- msg = " Failed to connect endpoints "
+ msg = "Failed to connect endpoints "
if proc.poll():
self.error(msg, out, err)
parser.add_option("-f", "--vif-name-file", dest="vif_name_file",
help = "File to store the virtual interface name assigned by the OS",
- default = "if_name", type="str")
+ default = "vif_name", type="str")
parser.add_option("-S", "--socket-name", dest="socket_name",
help = "Name for the unix socket used to interact with this process",
vsys.vif_up(vif_name, ip4_address, net_prefix, snat = snat,
pointopoint = pointopoint, txqueuelen = txqueuelen)
- # Saving interface name to 'if_name_file
+ # Saving interface name to vif_name_file
f = open(vif_name_file, 'w')
f.write(vif_name)
f.close()
def get_options():
usage = ("usage: %prog -N <vif-name> -t <vif-type> -a <ip4-address> "
- "-n <net-prefix> -s <snat> -p <pointopoint> "
- "-q <txqueuelen> -g <gre_key> -G <gre_remote> ")
+ "-n <net-prefix> -s <snat> -p <pointopoint> -q <txqueuelen> "
+ "-g <gre_key> -G <gre_remote> -f <vif-name-file> ")
parser = OptionParser(usage = usage)
default = None,
type="str")
+ parser.add_option("-f", "--vif-name-file", dest="vif_name_file",
+ help = "File to store the virtual interface name assigned by the OS",
+ default = "vif_name", type="str")
+
(options, args) = parser.parse_args()
vif_type = vsys.IFF_TAP
return (options.vif_name, vif_type, options.ip4_address,
options.net_prefix, options.snat, options.pointopoint,
- options.txqueuelen, options.gre_key, options.gre_remote)
+ options.txqueuelen, options.gre_key, options.gre_remote,
+ options.vif_name_file)
if __name__ == '__main__':
(vif_name, vif_type, ip4_address, net_prefix, snat, pointopoint,
- txqueuelen, gre_key, gre_remote) = get_options()
+ txqueuelen, gre_key, gre_remote, vif_name_file) = get_options()
if (gre_key):
import pwd
vsys.vif_up(vif_name, ip4_address, net_prefix, snat = snat,
pointopoint = pointopoint, txqueuelen = txqueuelen,
gre_key = gre_key, gre_remote = gre_remote)
-
+
+ # Saving interface name to vif_name_file
+ f = open(vif_name_file, 'w')
+ f.write(vif_name)
+ f.close()
+
+
pointopoint = Attribute("pointopoint", "Peer IP address",
flags = Flags.Design)
- tear_down = Attribute("tearDown", "Bash script to be executed before " + \
- "releasing the resource",
+ txqueuelen = Attribute("txqueuelen", "Length of transmission queue",
+ flags = Flags.Design)
+
+ txqueuelen = Attribute("txqueuelen", "Length of transmission queue",
+ flags = Flags.Design)
+
+ gre_key = Attribute("greKey",
+ "GRE key to be used to configure GRE tunnel",
+ default = "1",
+ flags = Flags.Design)
+
+ gre_remote = Attribute("greRemote",
+ "Public IP of remote endpoint for GRE tunnel",
+ flags = Flags.Design)
+
+ tear_down = Attribute("tearDown",
+ "Bash script to be executed before releasing the resource",
flags = Flags.Design)
cls._register_attribute(ip4)
cls._register_attribute(up)
cls._register_attribute(snat)
cls._register_attribute(pointopoint)
+ cls._register_attribute(txqueuelen)
+ cls._register_attribute(gre_key)
+ cls._register_attribute(gre_remote)
cls._register_attribute(tear_down)
def __init__(self, ec, guid):
super(PlanetlabTap, self).__init__(ec, guid)
self._home = "tap-%s" % self.guid
+ self._gre_enabled = False
@property
def node(self):
if node: return node[0]
return None
+ @property
+ def gre_enabled(self):
+ if not self._gre_enabled:
+ from nepi.resources.linux.gretunnel import LinuxGRETunnel
+ gre = self.get_connected(LinuxGRETunnel.get_rtype())
+ if gre: self._gre_enabled = True
+
+ return self._gre_enabled
+
def upload_sources(self):
scripts = []
os.path.join(self.app_home, "stop.sh"),
text = True,
# Overwrite file every time.
- # The stop.sh has the path to the socket, wich should change
+ # The stop.sh has the path to the socket, which should change
# on every experiment run.
overwrite = True)
def upload_start_command(self):
- # Overwrite file every time.
- # The start.sh has the path to the socket, wich should change
- # on every experiment run.
- super(PlanetlabTap, self).upload_start_command(overwrite = True)
-
- # We want to make sure the device is up and running
- # before the deploy finishes, so we execute now the
- # start script. We run it in background, because the
- # TAP will live for as long as the process that
- # created it is running, and wait until the TAP
- # is created.
- self._run_in_background()
-
- # After creating the TAP, the pl-vif-create.py script
- # will write the name of the TAP to a file. We wait until
- # we can read the interface name from the file.
- vif_name = self.wait_vif_name()
- self.set("deviceName", vif_name)
+ # If GRE mode is enabled, TAP creation is delayed until the
+ # tunnel is established
+ if not self.gre_enabled:
+ # Overwrite file every time.
+ # The start.sh has the path to the socket, wich should change
+ # on every experiment run.
+ super(PlanetlabTap, self).upload_start_command(overwrite = True)
+
+ # We want to make sure the device is up and running
+ # before the deploy finishes, so we execute now the
+ # start script. We run it in background, because the
+ # TAP will live for as long as the process that
+ # created it is running, and wait until the TAP
+ # is created.
+ self._run_in_background()
+
+ # After creating the TAP, the pl-vif-create.py script
+ # will write the name of the TAP to a file. We wait until
+ # we can read the interface name from the file.
+ vif_name = self.wait_vif_name()
+ self.set("deviceName", vif_name)
def do_deploy(self):
if not self.node or self.node.state < ResourceState.PROVISIONED:
@property
def state(self):
- # First check if the ccnd has failed
state_check_delay = 0.5
if self._state == ResourceState.STARTED and \
tdiffsec(tnow(), self._last_state_check) > state_check_delay:
def udp_connect_command(self, remote_endpoint, connection_run_home,
cipher, cipher_key, bwlimit, txqueuelen):
- # Generate UDP connect command
+ # Set the remote endpoint
+ self.set("pointopoint", remote_endpoint.get("ip4"))
+
remote_ip = socket.gethostbyname(
remote_endpoint.node.get("hostname"))
ret_file = os.path.join(connection_run_home,
"ret_file")
+ # Generate UDP connect command
# Use pl-vif-up.py script to configure TAP with peer info
- command = ["( sudo -S "]
- command.append("PYTHONPATH=$PYTHONPATH:${SRC}")
- command.append("python ${SRC}/pl-vif-up.py")
- command.append("-N %s" % self.get("deviceName"))
- command.append("-t %s" % self.vif_type)
- command.append("-a %s" % self.get("ip4"))
- command.append("-n %d" % self.get("prefix4"))
- if self.get("snat") == True:
- command.append("-s")
- command.append("-p %s" % remote_endpoint.get("ip4"))
+ vif_up_command = self._vif_up_command
+
+ command = ["( "]
+ command.append(vif_up_command)
# Use pl-vid-udp-connect.py to stablish the tunnel between endpoints
command.append(") & (")
return command
+ def gre_connect_command(self, remote_endpoint, connection_run_home):
+ # Set the remote endpoint
+ self.set("pointopoint", remote_endpoint.get("ip4"))
+ self.set("gre_remote", socket.gethostbyname(
+ remote_endpoint.node.get("hostname")))
+
+ # Generate GRE connect command
+
+ # Use vif_down command to first kill existing TAP in GRE mode
+ vif_down_command = self._vif_command_command
+
+ # Use pl-vif-up.py script to configure TAP with peer info
+ vif_up_command = self._vif_up_command
+
+ command = ["("]
+ command.append(vif_down_command)
+ command.append(") ; (")
+ command.append(vif_up_command)
+ command.append(")")
+
+ command = " ".join(command)
+ command = self.replace_paths(command)
+
+ return command
+
@property
def _start_command(self):
- command = ["sudo -S python ${SRC}/pl-vif-create.py"]
-
+ if self.gre_enabled:
+ command = []
+ else:
+ command = ["sudo -S python ${SRC}/pl-vif-create.py"]
+
+ command.append("-t %s" % self.vif_type)
+ command.append("-a %s" % self.get("ip4"))
+ command.append("-n %d" % self.get("prefix4"))
+ command.append("-f %s " % self.vif_name_file)
+ command.append("-S %s " % self.sock_name)
+
+ if self.get("snat") == True:
+ command.append("-s")
+
+ if self.get("pointopoint"):
+ command.append("-p %s" % self.get("pointopoint"))
+
+ if self.get("txqueuelen"):
+ command.append("-q %s" % self.get("txqueuelen"))
+
+ return " ".join(command)
+
+ @property
+ def _stop_command(self):
+ if self.gre_enabled:
+ command = self._vif_down_command()
+ else:
+ command = ["sudo -S "]
+ command.append("PYTHONPATH=$PYTHONPATH:${SRC}")
+ command.append("python ${SRC}/pl-vif-down.py")
+ command.append("-S %s " % self.sock_name)
+ command = " ".join(command)
+
+ return command
+
+ @property
+ def _vif_up_command(self):
+ if self.gre_enabled:
+ device_name = "%s" % self.guid
+ else:
+ device_name = self.get("deviceName")
+
+ # Use pl-vif-up.py script to configure TAP
+ command = ["sudo -S "]
+ command.append("PYTHONPATH=$PYTHONPATH:${SRC}")
+ command.append("python ${SRC}/pl-vif-up.py")
+ command.append("-N %s" % self.get("deviceName"))
command.append("-t %s" % self.vif_type)
command.append("-a %s" % self.get("ip4"))
command.append("-n %d" % self.get("prefix4"))
- command.append("-f %s " % self.vif_name_file)
- command.append("-S %s " % self.sock_name)
if self.get("snat") == True:
command.append("-s")
if self.get("pointopoint"):
command.append("-p %s" % self.get("pointopoint"))
+
+ if self.get("txqueuelen"):
+ command.append("-q %s" % self.get("txqueuelen"))
+
+ if self.gre_enabled:
+ command.append("-g %s" % self.gre("greKey"))
+ command.append("-G %s" % self.gre("greRemote"))
+
+ command.append("-f %s " % self.vif_name_file)
return " ".join(command)
@property
- def _stop_command(self):
- command = ["sudo -S python ${SRC}/pl-vif-down.py"]
-
- command.append("-S %s " % self.sock_name)
+ def _vif_down_command(self):
+ command = ["sudo -S "]
+ command.append("PYTHONPATH=$PYTHONPATH:${SRC}")
+ command.append("python ${SRC}/pl-vif-down.py")
+ command.append("-D")
+ command.append("-N %s " % self.get("deviceName"))
+
return " ".join(command)
@property
class PlanetlabTapTestCase(unittest.TestCase):
def setUp(self):
- self.host = "nepi5.pl.sophia.inria.fr"
+ self.host = "planetlab1.informatik.uni-erlangen.de"
self.user = "inria_nepi"
+ self.identity = "%s/.ssh/id_rsa_planetlab" % (os.environ['HOME'])
+ self.netblock = "192.168.3"
+ #self.host = "nepi2.pl.sophia.inria.fr"
+ #self.user = "inria_nepi"
+ #self.identity = None
+ #self.netblock = "192.168.1"
@skipIfNotAlive
- def t_tap_create(self, host, user):
+ def t_tap_create(self, host, user, identity):
ec = ExperimentController(exp_id = "test-tap-create")
node = ec.register_resource("PlanetlabNode")
ec.set(node, "hostname", host)
ec.set(node, "username", user)
+ ec.set(node, "identity", identity)
ec.set(node, "cleanHome", True)
ec.set(node, "cleanProcesses", True)
tap = ec.register_resource("PlanetlabTap")
- ec.set(tap, "ip4", "192.168.1.1")
+ ec.set(tap, "ip4", "%s.1" % self.netblock)
ec.set(tap, "prefix4", 24)
ec.register_connection(tap, node)
app = ec.register_resource("LinuxApplication")
- cmd = "ping -c3 192.168.1.1"
+ cmd = "ping -c3 %s.1" % self.netblock
ec.set(app, "command", cmd)
ec.register_connection(app, node)
ec.shutdown()
def test_tap_create(self):
- self.t_tap_create(self.host, self.user)
+ self.t_tap_create(self.host, self.user, self.identity)
if __name__ == '__main__':
unittest.main()
class PlanetlabTunTestCase(unittest.TestCase):
def setUp(self):
- self.host = "nepi2.pl.sophia.inria.fr"
+ self.host = "planetlab1.informatik.uni-erlangen.de"
self.user = "inria_nepi"
+ self.identity = "%s/.ssh/id_rsa_planetlab" % (os.environ['HOME'])
+ self.netblock = "192.168.3"
+ #self.host = "nepi2.pl.sophia.inria.fr"
+ #self.user = "inria_nepi"
+ #self.identity = None
+ #self.netblock = "192.168.1"
@skipIfNotAlive
- def t_tun_create(self, host, user):
+ def t_tun_create(self, host, user, identity):
ec = ExperimentController(exp_id = "test-un-create")
node = ec.register_resource("PlanetlabNode")
ec.set(node, "hostname", host)
ec.set(node, "username", user)
+ ec.set(node, "identity", identity)
ec.set(node, "cleanHome", True)
ec.set(node, "cleanProcesses", True)
tun = ec.register_resource("PlanetlabTun")
- ec.set(tun, "ip4", "192.168.1.1")
+ ec.set(tun, "ip4", "%s.1" % self.netblock)
ec.set(tun, "prefix4", 24)
ec.register_connection(tun, node)
app = ec.register_resource("LinuxApplication")
- cmd = "ping -c3 192.168.1.1"
+ cmd = "ping -c3 %s.1" % self.netblock
ec.set(app, "command", cmd)
ec.register_connection(app, node)
ec.shutdown()
def test_tun_create(self):
- self.t_tun_create(self.host, self.user)
+ self.t_tun_create(self.host, self.user, self.identity)
if __name__ == '__main__':
unittest.main()
git://git.onelab.eu / nepi.git / commitdiff