src/nepi/resources/linux/udptunnel.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program is distributed in the hope that it will be useful,
  11 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19
  20 from nepi.execution.attribute import Attribute, Flags, Types
  21 from nepi.execution.resource import clsinit_copy, ResourceState, \
  22         reschedule_delay
  23 from nepi.resources.linux.application import LinuxApplication
  24 from nepi.util.sshfuncs import ProcStatus
  25 from nepi.util.timefuncs import tnow, tdiffsec
  26
  27 import os
  28 import socket
  29 import time
  30
  31 @clsinit_copy
  32 class UdpTunnel(LinuxApplication):
  33     _rtype = "UdpTunnel"
  34     _help = "Constructs a tunnel between two Linux endpoints using a UDP connection "
  35     _backend = "linux"
  36
  37     @classmethod
  38     def _register_attributes(cls):
  39         cipher = Attribute("cipher",
  40                "Cipher to encript communication. "
  41                 "One of PLAIN, AES, Blowfish, DES, DES3. ",
  42                 default = None,
  43                 allowed = ["PLAIN", "AES", "Blowfish", "DES", "DES3"],
  44                 type = Types.Enumerate,
  45                 flags = Flags.Design)
  46
  47         cipher_key = Attribute("cipherKey",
  48                 "Specify a symmetric encryption key with which to protect "
  49                 "packets across the tunnel. python-crypto must be installed "
  50                 "on the system." ,
  51                 flags = Flags.Design)
  52
  53         txqueuelen = Attribute("txQueueLen",
  54                 "Specifies the interface's transmission queue length. "
  55                 "Defaults to 1000. ",
  56                 type = Types.Integer,
  57                 flags = Flags.Design)
  58
  59         bwlimit = Attribute("bwLimit",
  60                 "Specifies the interface's emulated bandwidth in bytes "
  61                 "per second.",
  62                 type = Types.Integer,
  63                 flags = Flags.Design)
  64
  65         cls._register_attribute(cipher)
  66         cls._register_attribute(cipher_key)
  67         cls._register_attribute(txqueuelen)
  68         cls._register_attribute(bwlimit)
  69
  70     def __init__(self, ec, guid):
  71         super(UdpTunnel, self).__init__(ec, guid)
  72         self._home = "udp-tunnel-%s" % self.guid
  73         self._pid1 = None
  74         self._ppid1 = None
  75         self._pid2 = None
  76         self._ppid2 = None
  77
  78     def log_message(self, msg):
  79         return " guid %d - tunnel %s - %s - %s " % (self.guid,
  80                 self.endpoint1.node.get("hostname"),
  81                 self.endpoint2.node.get("hostname"),
  82                 msg)
  83
  84     def get_endpoints(self):
  85         """ Returns the list of RM that are endpoints to the tunnel
  86         """
  87         connected = []
  88         for guid in self.connections:
  89             rm = self.ec.get_resource(guid)
  90             if hasattr(rm, "udp_connect_command"):
  91                 connected.append(rm)
  92         return connected
  93
  94     @property
  95     def endpoint1(self):
  96         endpoints = self.get_endpoints()
  97         if endpoints: return endpoints[0]
  98         return None
  99
 100     @property
 101     def endpoint2(self):
 102         endpoints = self.get_endpoints()
 103         if endpoints and len(endpoints) > 1: return endpoints[1]
 104         return None
 105
 106     def app_home(self, endpoint):
 107         return os.path.join(endpoint.node.exp_home, self._home)
 108
 109     def run_home(self, endpoint):
 110         return os.path.join(self.app_home(endpoint), self.ec.run_id)
 111
 112     def udp_connect(self, endpoint, remote_ip):
 113         # Get udp connect command
 114         local_port_file = os.path.join(self.run_home(endpoint),
 115                 "local_port")
 116         remote_port_file = os.path.join(self.run_home(endpoint),
 117                 "remote_port")
 118         ret_file = os.path.join(self.run_home(endpoint),
 119                 "ret_file")
 120         cipher = self.get("cipher")
 121         cipher_key = self.get("cipherKey")
 122         bwlimit = self.get("bwLimit")
 123         txqueuelen = self.get("txQueueLen")
 124         udp_connect_command = endpoint.udp_connect_command(
 125                 remote_ip, local_port_file, remote_port_file,
 126                 ret_file, cipher, cipher_key, bwlimit, txqueuelen)
 127
 128         # upload command to connect.sh script
 129         shfile = os.path.join(self.app_home(endpoint), "udp-connect.sh")
 130         endpoint.node.upload(udp_connect_command,
 131                 shfile,
 132                 text = True,
 133                 overwrite = False)
 134
 135         # invoke connect script
 136         cmd = "bash %s" % shfile
 137         (out, err), proc = endpoint.node.run(cmd, self.run_home(endpoint))
 138
 139         # check if execution errors occurred
 140         msg = " Failed to connect endpoints "
 141
 142         if proc.poll():
 143             self.error(msg, out, err)
 144             raise RuntimeError, msg
 145
 146         # Wait for pid file to be generated
 147         pid, ppid = endpoint.node.wait_pid(self.run_home(endpoint))
 148
 149         # If the process is not running, check for error information
 150         # on the remote machine
 151         if not pid or not ppid:
 152             (out, err), proc = endpoint.node.check_errors(self.run_home(endpoint))
 153             # Out is what was written in the stderr file
 154             if err:
 155                 msg = " Failed to start command '%s' " % command
 156                 self.error(msg, out, err)
 157                 raise RuntimeError, msg
 158
 159         # wait until port is written to file
 160         port = self.wait_local_port(endpoint)
 161         return (port, pid, ppid)
 162
 163     def do_provision(self):
 164         # create run dir for tunnel on each node
 165         self.endpoint1.node.mkdir(self.run_home(self.endpoint1))
 166         self.endpoint2.node.mkdir(self.run_home(self.endpoint2))
 167
 168         # Invoke connect script in endpoint 1
 169         remote_ip1 = socket.gethostbyname(self.endpoint2.node.get("hostname"))
 170         (port1, self._pid1, self._ppid1) = self.udp_connect(self.endpoint1,
 171                 remote_ip1)
 172
 173         # Invoke connect script in endpoint 2
 174         remote_ip2 = socket.gethostbyname(self.endpoint1.node.get("hostname"))
 175         (port2, self._pid2, self._ppid2) = self.udp_connect(self.endpoint2,
 176                 remote_ip2)
 177
 178         # upload file with port 2 to endpoint 1
 179         self.upload_remote_port(self.endpoint1, port2)
 180
 181         # upload file with port 1 to endpoint 2
 182         self.upload_remote_port(self.endpoint2, port1)
 183
 184         # check if connection was successful on both sides
 185         self.wait_result(self.endpoint1)
 186         self.wait_result(self.endpoint2)
 187
 188         self.info("Provisioning finished")
 189
 190         self.set_provisioned()
 191
 192     def do_deploy(self):
 193         if (not self.endpoint1 or self.endpoint1.state < ResourceState.READY) or \
 194             (not self.endpoint2 or self.endpoint2.state < ResourceState.READY):
 195             self.ec.schedule(reschedule_delay, self.deploy)
 196         else:
 197             self.do_discover()
 198             self.do_provision()
 199
 200             self.set_ready()
 201
 202     def do_start(self):
 203         if self.state == ResourceState.READY:
 204             command = self.get("command")
 205             self.info("Starting command '%s'" % command)
 206
 207             self.set_started()
 208         else:
 209             msg = " Failed to execute command '%s'" % command
 210             self.error(msg, out, err)
 211             raise RuntimeError, msg
 212
 213     def do_stop(self):
 214         """ Stops application execution
 215         """
 216         if self.state == ResourceState.STARTED:
 217             self.info("Stopping tunnel")
 218
 219             # Only try to kill the process if the pid and ppid
 220             # were retrieved
 221             if self._pid1 and self._ppid1 and self._pid2 and self._ppid2:
 222                 (out1, err1), proc1 = self.endpoint1.node.kill(self._pid1,
 223                         self._ppid1, sudo = True)
 224                 (out2, err2), proc2 = self.endpoint2.node.kill(self._pid2,
 225                         self._ppid2, sudo = True)
 226
 227                 if (proc1.poll() and err1) or (proc2.poll() and err2):
 228                     # check if execution errors occurred
 229                     msg = " Failed to STOP tunnel"
 230                     self.error(msg, err1, err2)
 231                     raise RuntimeError, msg
 232
 233             self.set_stopped()
 234
 235     @property
 236     def state(self):
 237         """ Returns the state of the application
 238         """
 239         if self._state == ResourceState.STARTED:
 240             # In order to avoid overwhelming the remote host and
 241             # the local processor with too many ssh queries, the state is only
 242             # requested every 'state_check_delay' seconds.
 243             state_check_delay = 0.5
 244             if tdiffsec(tnow(), self._last_state_check) > state_check_delay:
 245                 if self._pid1 and self._ppid1 and self._pid2 and self._ppid2:
 246                     # Make sure the process is still running in background
 247                     # No execution errors occurred. Make sure the background
 248                     # process with the recorded pid is still running.
 249                     status1 = self.endpoint1.node.status(self._pid1, self._ppid1)
 250                     status2 = self.endpoint2.node.status(self._pid2, self._ppid2)
 251
 252                     if status1 == ProcStatus.FINISHED and \
 253                             status2 == ProcStatus.FINISHED:
 254
 255                         # check if execution errors occurred
 256                         (out1, err1), proc1 = self.endpoint1.node.check_errors(
 257                                 self.run_home(self.endpoint1))
 258
 259                         (out2, err2), proc2 = self.endpoint2.node.check_errors(
 260                                 self.run_home(self.endpoint2))
 261
 262                         if err1 or err2:
 263                             msg = "Error occurred in tunnel"
 264                             self.error(msg, err1, err2)
 265                             self.fail()
 266                         else:
 267                             self.set_stopped()
 268
 269                 self._last_state_check = tnow()
 270
 271         return self._state
 272
 273     def wait_local_port(self, endpoint):
 274         """ Waits until the local_port file for the endpoint is generated,
 275         and returns the port number
 276
 277         """
 278         return self.wait_file(endpoint, "local_port")
 279
 280     def wait_result(self, endpoint):
 281         """ Waits until the return code file for the endpoint is generated
 282
 283         """
 284         return self.wait_file(endpoint, "ret_file")
 285
 286     def wait_file(self, endpoint, filename):
 287         """ Waits until file on endpoint is generated """
 288         result = None
 289         delay = 1.0
 290
 291         for i in xrange(20):
 292             (out, err), proc = endpoint.node.check_output(
 293                     self.run_home(endpoint), filename)
 294
 295             if out:
 296                 result = out.strip()
 297                 break
 298             else:
 299                 time.sleep(delay)
 300                 delay = delay * 1.5
 301         else:
 302             msg = "Couldn't retrieve %s" % filename
 303             self.error(msg, out, err)
 304             raise RuntimeError, msg
 305
 306         return result
 307
 308     def upload_remote_port(self, endpoint, port):
 309         # upload remote port number to file
 310         port = "%s\n" % port
 311         endpoint.node.upload(port,
 312                 os.path.join(self.run_home(endpoint), "remote_port"),
 313                 text = True,
 314                 overwrite = False)
 315
 316     def valid_connection(self, guid):
 317         # TODO: Validate!
 318         return True
 319