src/nepi/resources/omf/wilabt_node.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 version 2 as
   7 #    published by the Free Software Foundation;
   8 #
   9 #    This program is distributed in the hope that it will be useful,
  10 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 #    GNU General Public License for more details.
  13 #
  14 #    You should have received a copy of the GNU General Public License
  15 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17 # Author: Lucia Guevgeozian <lucia.guevgeozian_odizzio@inria.fr>
  18
  19 from nepi.execution.attribute import Attribute, Flags, Types
  20 from nepi.execution.resource import ResourceManager, clsinit_copy, \
  21         ResourceState
  22 from nepi.resources.omf.node import OMFNode
  23 from nepi.util.sfaapi import SFAAPIFactory
  24 from nepi.util.execfuncs import lexec
  25 from nepi.util import sshfuncs
  26
  27 from random import randint
  28 import time
  29 import re
  30 import weakref
  31 import socket
  32 import threading
  33 import datetime
  34
  35 @clsinit_copy
  36 class WilabtSfaNode(OMFNode):
  37     _rtype = "wilabt::sfa::Node"
  38     _help = "Controls a Wilabt host accessible using a SSH key " \
  39             "and provisioned using SFA"
  40     _platform = "omf"
  41
  42     @classmethod
  43     def _register_attributes(cls):
  44
  45         username = Attribute("username", "Local account username",
  46                 flags = Flags.Credential)
  47
  48         identity = Attribute("identity", "SSH identity file",
  49                 flags = Flags.Credential)
  50
  51         server_key = Attribute("serverKey", "Server public key",
  52                 flags = Flags.Design)
  53
  54         sfa_user = Attribute("sfauser", "SFA user",
  55                     flags = Flags.Credential)
  56
  57         sfa_private_key = Attribute("sfaPrivateKey", "SFA path to the private key \
  58                             used to generate the user credential",
  59                             flags = Flags.Credential)
  60
  61         slicename = Attribute("slicename", "SFA slice for the experiment",
  62                     flags = Flags.Credential)
  63
  64         gateway_user = Attribute("gatewayUser", "Gateway account username",
  65                 flags = Flags.Design)
  66
  67         gateway = Attribute("gateway", "Hostname of the gateway machine",
  68                 flags = Flags.Design)
  69
  70         host = Attribute("host", "Name of the physical machine",
  71                 flags = Flags.Design)
  72
  73         disk_image = Attribute("disk_image", "Specify a specific disk image for a node",
  74                 flags = Flags.Design)
  75
  76         cls._register_attribute(username)
  77         cls._register_attribute(identity)
  78         cls._register_attribute(server_key)
  79         cls._register_attribute(sfa_user)
  80         cls._register_attribute(sfa_private_key)
  81         cls._register_attribute(slicename)
  82         cls._register_attribute(gateway_user)
  83         cls._register_attribute(gateway)
  84         cls._register_attribute(host)
  85         cls._register_attribute(disk_image)
  86
  87     def __init__(self, ec, guid):
  88         super(WilabtSfaNode, self).__init__(ec, guid)
  89
  90         self._ecobj = weakref.ref(ec)
  91         self._sfaapi = None
  92         self._node_to_provision = None
  93         self._slicenode = False
  94         self._host = False
  95         self._username = None
  96
  97     def _skip_provision(self):
  98         sfa_user = self.get("sfauser")
  99         if not sfa_user:
 100             return True
 101         else: return False
 102
 103     @property
 104     def sfaapi(self):
 105         """
 106         Property to instanciate the SFA API based in sfi client.
 107         For each SFA method called this instance is used.
 108         """
 109         if not self._sfaapi:
 110             sfa_user = self.get("sfauser")
 111             sfa_sm = "http://www.wilab2.ilabt.iminds.be:12369/protogeni/xmlrpc/am/3.0"
 112             sfa_auth = '.'.join(sfa_user.split('.')[:2])
 113             sfa_registry = "http://sfa3.planet-lab.eu:12345/"
 114             sfa_private_key = self.get("sfaPrivateKey")
 115             batch = True
 116
 117             _sfaapi = SFAAPIFactory.get_api(sfa_user, sfa_auth,
 118                 sfa_registry, sfa_sm, sfa_private_key, self._ecobj(), batch, WilabtSfaNode._rtype)
 119
 120             if not _sfaapi:
 121                 self.fail_sfaapi()
 122
 123             self._sfaapi = weakref.ref(_sfaapi)
 124
 125         return self._sfaapi()
 126
 127     def do_discover(self):
 128         """
 129         Based on the attributes defined by the user, discover the suitable
 130         node for provision.
 131         """
 132         nodes = self.sfaapi.get_resources_hrn()
 133
 134         host = self._get_host()
 135         if host:
 136             # the user specified one particular node to be provisioned
 137             self._host = True
 138             host_hrn = nodes[host]
 139
 140             # check that the node is not blacklisted or being provisioned
 141             # by other RM
 142             if not self._blacklisted(host_hrn):
 143                 if not self._reserved(host_hrn):
 144                     if self._check_if_in_slice([host_hrn]):
 145                         self.debug("Node already in slice %s" % host_hrn)
 146                         self._slicenode = True
 147                     host = host + '.wilab2.ilabt.iminds.be'
 148                     self.set('host', host)
 149                     self._node_to_provision = host_hrn
 150                     super(WilabtSfaNode, self).do_discover()
 151
 152     def do_provision(self):
 153         """
 154         Add node to user's slice and verifing that the node is functioning
 155         correctly. Check ssh, omf rc running, hostname, file system.
 156         """
 157         provision_ok = False
 158         ssh_ok = False
 159         proc_ok = False
 160         timeout = 300
 161
 162         while not provision_ok:
 163             node = self._node_to_provision
 164             #if self._slicenode:
 165             #    self._delete_from_slice()
 166             #    self.debug("Waiting 480 sec for re-adding to slice")
 167             #    time.sleep(480) # Timout for the testbed to allow a new reservation
 168             self._add_node_to_slice(node)
 169             t = 0
 170             while not self._check_if_in_slice([node]) and t < timeout \
 171                 and not self._ecobj().abort:
 172                 t = t + 5
 173                 time.sleep(t)
 174                 self.debug("Waiting 5 sec for resources to be added")
 175                 continue
 176
 177             if not self._check_if_in_slice([node]):
 178                 self.debug("Couldn't add node %s to slice" % node)
 179                 self.fail_node_not_available(node)
 180
 181             self._get_username()
 182             ssh_ok = self._check_ssh_loop()
 183
 184             if not ssh_ok:
 185                 # the timeout was reach without establishing ssh connection
 186                 # the node is blacklisted, and a new
 187                 # node to provision is discovered
 188                 self._blacklist_node(node)
 189                 self.do_discover()
 190                 continue
 191
 192             # check /proc directory is mounted (ssh_ok = True)
 193             # file system is not read only, hostname is correct
 194             # and omf_rc process is up
 195             else:
 196                 if not self._check_fs():
 197                     self.do_discover()
 198                     continue
 199                 if not self._check_omfrc():
 200                     self.do_discover()
 201                     continue
 202                 if not self._check_hostname():
 203                     self.do_discover()
 204                     continue
 205
 206                 else:
 207                     provision_ok = True
 208                     if not self.get('host'):
 209                         self._set_host_attr(node)
 210                     self.info(" Node provisioned ")
 211
 212         super(WilabtSfaNode, self).do_provision()
 213
 214     def do_deploy(self):
 215         if self.state == ResourceState.NEW:
 216             self.info("Deploying w-iLab.t node")
 217             self.do_discover()
 218             self.do_provision()
 219         super(WilabtSfaNode, self).do_deploy()
 220
 221     def do_release(self):
 222         super(WilabtSfaNode, self).do_release()
 223         if self.state == ResourceState.RELEASED and not self._skip_provision():
 224             self.debug(" Releasing SFA API ")
 225             self.sfaapi.release()
 226
 227     def _blacklisted(self, host_hrn):
 228         """
 229         Check in the SFA API that the node is not in the blacklist.
 230         """
 231         if self.sfaapi.blacklisted(host_hrn):
 232            self.fail_node_not_available(host_hrn)
 233         return False
 234
 235     def _reserved(self, host_hrn):
 236         """
 237         Check in the SFA API that the node is not in the reserved
 238         list.
 239         """
 240         if self.sfaapi.reserved(host_hrn):
 241             self.fail_node_not_available(host_hrn)
 242         return False
 243
 244     def _get_username(self):
 245         """
 246         Get the username for login in to the nodes from RSpec.
 247         Wilabt username is not made out of any convention, it
 248         has to be retrived from the manifest RSpec.
 249         """
 250         slicename = self.get("slicename")
 251         if self._username is None:
 252             slice_info = self.sfaapi.get_slice_resources(slicename)
 253             username = slice_info['resource'][0]['services'][0]['login'][0]['username']
 254             self.set('username', username)
 255             self.debug("Retriving username information from RSpec %s" % username)
 256             self._username = username
 257
 258     def _check_ssh_loop(self):
 259         """
 260         Check that the ssh login is possible. In wilabt is done
 261         through the gateway because is private testbed.
 262         """
 263         t = 0
 264         timeout = 1200
 265         ssh_ok = False
 266         while t < timeout and not ssh_ok:
 267             cmd = 'echo \'GOOD NODE\''
 268             ((out, err), proc) = self.execute(cmd)
 269             if out.find("GOOD NODE") < 0:
 270                 self.debug( "No SSH connection, waiting 20s" )
 271                 t = t + 20
 272                 time.sleep(20)
 273                 continue
 274             else:
 275                 self.debug( "SSH OK" )
 276                 ssh_ok = True
 277                 continue
 278         return ssh_ok
 279
 280     def _check_fs(self):
 281         """
 282         Check file system, /proc well mounted.
 283         """
 284         cmd = 'mount |grep proc'
 285         ((out, err), proc) = self.execute(cmd)
 286         if out.find("/proc type proc") < 0:
 287             self.warning(" Corrupted file system ")
 288             self._blacklist_node(node)
 289             return False
 290         return True
 291
 292     def _check_omfrc(self):
 293         """
 294         Check that OMF 6 resource controller is running.
 295         """
 296         cmd = 'ps aux|grep omf'
 297         ((out, err), proc) = self.execute(cmd)
 298         if out.find("/usr/local/rvm/gems/ruby-1.9.3-p286@omf/bin/omf_rc") < 0:
 299             return False
 300         return True
 301
 302     def _check_hostname(self):
 303         """
 304         Check that the hostname in the image is not set to localhost.
 305         """
 306         cmd = 'hostname'
 307         ((out, err), proc) = self.execute(cmd)
 308         if 'localhost' in out.lower():
 309             return False
 310         else:
 311             self.set('hostname', out.strip())
 312         return True
 313
 314     def execute(self, command,
 315         sudo = False,
 316         env = None,
 317         tty = False,
 318         forward_x11 = False,
 319         retry = 3,
 320         connect_timeout = 30,
 321         strict_host_checking = False,
 322         persistent = True,
 323         blocking = True,
 324         ):
 325         """ Notice that this invocation will block until the
 326         execution finishes. If this is not the desired behavior,
 327         use 'run' instead."""
 328         (out, err), proc = sshfuncs.rexec(
 329             command,
 330             host = self.get("host"),
 331             user = self.get("username"),
 332             port = 22,
 333             gwuser = self.get("gatewayUser"),
 334             gw = self.get("gateway"),
 335             agent = True,
 336             sudo = sudo,
 337             identity = self.get("identity"),
 338             server_key = self.get("serverKey"),
 339             env = env,
 340             tty = tty,
 341             forward_x11 = forward_x11,
 342             retry = retry,
 343             connect_timeout = connect_timeout,
 344             persistent = persistent,
 345             blocking = blocking,
 346             strict_host_checking = strict_host_checking
 347             )
 348
 349         return (out, err), proc
 350
 351
 352     def _add_node_to_slice(self, host_hrn):
 353         """
 354         Add node to slice, using SFA API. Actually Wilabt testbed
 355         doesn't allow adding nodes, in fact in the API there is method
 356         to group all the nodes instanciated as WilabtSfaNodes and the
 357         Allocate and Provision is done with the last call at
 358         sfaapi.add_resource_to_slice_batch.
 359         """
 360         self.info(" Adding node to slice ")
 361         slicename = self.get("slicename")
 362         disk_image = self.get("disk_image")
 363         if disk_image is not None:
 364             properties = {'disk_image': disk_image}
 365         else: properties = None
 366         #properties = None
 367         self.sfaapi.add_resource_to_slice_batch(slicename, host_hrn, properties=properties)
 368
 369     def _delete_from_slice(self):
 370         """
 371         Delete every node from slice, using SFA API.
 372         Wilabt doesn't allow to remove one sliver so this method
 373         remove every slice from the slice.
 374         """
 375
 376         self.warning(" Deleting all slivers from slice ")
 377         slicename = self.get("slicename")
 378         self.sfaapi.remove_all_from_slice(slicename)
 379
 380     def _get_host(self):
 381         """
 382         Get the attribute hostname.
 383         """
 384         host = self.get("host")
 385         if host:
 386             return host
 387         else:
 388             return None
 389
 390     def _set_host_attr(self, node):
 391         """
 392         Query SFAAPI for the hostname of a certain host hrn and sets the
 393         attribute hostname, it will over write the previous value.
 394         """
 395         hosts_hrn = self.sfaapi.get_resources_hrn()
 396         for host, hrn  in hosts_hrn.iteritems():
 397             if hrn == node:
 398                 host = host + '.wilab2.ilabt.iminds.be'
 399                 self.set("host", host)
 400
 401     def _check_if_in_slice(self, hosts_hrn):
 402         """
 403         Check using SFA API if any host hrn from hosts_hrn is in the user's
 404         slice.
 405         """
 406         slicename = self.get("slicename")
 407         slice_nodes = self.sfaapi.get_slice_resources(slicename)['resource']
 408         if slice_nodes:
 409             if len(slice_nodes[0]['services']) != 0:
 410                 slice_nodes_hrn = self.sfaapi.get_resources_hrn(slice_nodes).values()
 411         else: slice_nodes_hrn = []
 412         nodes_inslice = list(set(hosts_hrn) & set(slice_nodes_hrn))
 413         return nodes_inslice
 414
 415     def _blacklist_node(self, host_hrn):
 416         """
 417         Add mal functioning node to blacklist (in SFA API).
 418         """
 419         self.warning(" Blacklisting malfunctioning node ")
 420         self.sfaapi.blacklist_resource(host_hrn)
 421         if not self._host:
 422             self.set('host', None)
 423         else:
 424             self.set('host', host_hrn.split('.').pop())
 425
 426     def _put_node_in_provision(self, host_hrn):
 427         """
 428         Add node to the list of nodes being provisioned, in order for other RMs
 429         to not try to provision the same one again.
 430         """
 431         self.sfaapi.reserve_resource(host_hrn)
 432
 433     def _get_ip(self, host):
 434         """
 435         Query cache for the IP of a node with certain hostname
 436         """
 437         try:
 438             ip = sshfuncs.gethostbyname(host)
 439         except:
 440             # Fail while trying to find the IP
 441             return None
 442         return ip
 443
 444     def fail_discovery(self):
 445         msg = "Discovery failed. No candidates found for node"
 446         self.error(msg)
 447         raise RuntimeError(msg)
 448
 449     def fail_node_not_alive(self, host=None):
 450         msg = "Node %s not alive" % host
 451         raise RuntimeError(msg)
 452
 453     def fail_node_not_available(self, host):
 454         msg = "Some nodes not available for provisioning"
 455         raise RuntimeError(msg)
 456
 457     def fail_not_enough_nodes(self):
 458         msg = "Not enough nodes available for provisioning"
 459         raise RuntimeError(msg)
 460
 461     def fail_sfaapi(self):
 462         msg = "Failing while trying to instanciate the SFA API."
 463         raise RuntimeError(msg)
 464
 465     def valid_connection(self, guid):
 466         # TODO: Validate!
 467         return True
 468
 469