src/nepi/resources/planetlab/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 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 #         Lucia Guevgeozian <lucia.guevgeozian_odizzio@inria.fr>
  20
  21 from nepi.execution.attribute import Attribute, Flags, Types
  22 from nepi.execution.resource import ResourceManager, clsinit_copy, ResourceState, \
  23         reschedule_delay
  24 from nepi.resources.linux.node import LinuxNode
  25 from nepi.resources.planetlab.plcapi import PLCAPIFactory
  26 from nepi.util.execfuncs import lexec
  27 from nepi.util import sshfuncs
  28
  29 from random import randint
  30 import time
  31 import threading
  32
  33 @clsinit_copy
  34 class PlanetlabNode(LinuxNode):
  35     _rtype = "PlanetlabNode"
  36     _help = "Controls a PlanetLab host accessible using a SSH key " \
  37             "associated to a PlanetLab user account"
  38     _backend = "planetlab"
  39
  40     lock = threading.Lock()
  41
  42     @classmethod
  43     def _register_attributes(cls):
  44         ip = Attribute("ip", "PlanetLab host public IP address",
  45                 flags = Flags.ReadOnly)
  46
  47         pl_url = Attribute("plcApiUrl", "URL of PlanetLab PLCAPI host \
  48                     (e.g. www.planet-lab.eu or www.planet-lab.org) ",
  49                     default = "www.planet-lab.eu",
  50                     flags = Flags.Credential)
  51
  52         pl_ptn = Attribute("plcApiPattern", "PLC API service regexp pattern \
  53                     (e.g. https://%(hostname)s:443/PLCAPI/ ) ",
  54                     default = "https://%(hostname)s:443/PLCAPI/",
  55                     flags = Flags.ExecReadOnly)
  56
  57         pl_user = Attribute("pluser", "PlanetLab account user, as the one to \
  58                     authenticate in the website) ",
  59                     flags = Flags.Credential)
  60
  61         pl_password = Attribute("plpassword",
  62                         "PlanetLab account password, as \
  63                         the one to authenticate in the website) ",
  64                         flags = Flags.Credential)
  65
  66         city = Attribute("city", "Constrain location (city) during resource \
  67                 discovery. May use wildcards.",
  68                 flags = Flags.Filter)
  69
  70         country = Attribute("country", "Constrain location (country) during \
  71                     resource discovery. May use wildcards.",
  72                     flags = Flags.Filter)
  73
  74         region = Attribute("region", "Constrain location (region) during \
  75                     resource discovery. May use wildcards.",
  76                     flags = Flags.Filter)
  77
  78         architecture = Attribute("architecture", "Constrain architecture \
  79                         during resource discovery.",
  80                         type = Types.Enumerate,
  81                         allowed = ["x86_64",
  82                                     "i386"],
  83                         flags = Flags.Filter)
  84
  85         operating_system = Attribute("operatingSystem", "Constrain operating \
  86                             system during resource discovery.",
  87                             type = Types.Enumerate,
  88                             allowed =  ["f8",
  89                                         "f12",
  90                                         "f14",
  91                                         "centos",
  92                                         "other"],
  93                             flags = Flags.Filter)
  94
  95         #site = Attribute("site", "Constrain the PlanetLab site this node \
  96         #        should reside on.",
  97         #        type = Types.Enumerate,
  98         #        allowed = ["PLE",
  99         #                    "PLC",
 100         #                    "PLJ"],
 101         #        flags = Flags.Filter)
 102
 103         min_reliability = Attribute("minReliability", "Constrain reliability \
 104                             while picking PlanetLab nodes. Specifies a lower \
 105                             acceptable bound.",
 106                             type = Types.Double,
 107                             range = (1, 100),
 108                             flags = Flags.Filter)
 109
 110         max_reliability = Attribute("maxReliability", "Constrain reliability \
 111                             while picking PlanetLab nodes. Specifies an upper \
 112                             acceptable bound.",
 113                             type = Types.Double,
 114                             range = (1, 100),
 115                             flags = Flags.Filter)
 116
 117         min_bandwidth = Attribute("minBandwidth", "Constrain available \
 118                             bandwidth while picking PlanetLab nodes. \
 119                             Specifies a lower acceptable bound.",
 120                             type = Types.Double,
 121                             range = (0, 2**31),
 122                             flags = Flags.Filter)
 123
 124         max_bandwidth = Attribute("maxBandwidth", "Constrain available \
 125                             bandwidth while picking PlanetLab nodes. \
 126                             Specifies an upper acceptable bound.",
 127                             type = Types.Double,
 128                             range = (0, 2**31),
 129                             flags = Flags.Filter)
 130
 131         min_load = Attribute("minLoad", "Constrain node load average while \
 132                     picking PlanetLab nodes. Specifies a lower acceptable \
 133                     bound.",
 134                     type = Types.Double,
 135                     range = (0, 2**31),
 136                     flags = Flags.Filter)
 137
 138         max_load = Attribute("maxLoad", "Constrain node load average while \
 139                     picking PlanetLab nodes. Specifies an upper acceptable \
 140                     bound.",
 141                     type = Types.Double,
 142                     range = (0, 2**31),
 143                     flags = Flags.Filter)
 144
 145         min_cpu = Attribute("minCpu", "Constrain available cpu time while \
 146                     picking PlanetLab nodes. Specifies a lower acceptable \
 147                     bound.",
 148                     type = Types.Double,
 149                     range = (0, 100),
 150                     flags = Flags.Filter)
 151
 152         max_cpu = Attribute("maxCpu", "Constrain available cpu time while \
 153                     picking PlanetLab nodes. Specifies an upper acceptable \
 154                     bound.",
 155                     type = Types.Double,
 156                     range = (0, 100),
 157                     flags = Flags.Filter)
 158
 159         timeframe = Attribute("timeframe", "Past time period in which to check\
 160                         information about the node. Values are year,month, \
 161                         week, latest",
 162                         default = "week",
 163                         type = Types.Enumerate,
 164                         allowed = ["latest",
 165                                     "week",
 166                                     "month",
 167                                     "year"],
 168                         flags = Flags.Filter)
 169
 170         cls._register_attribute(ip)
 171         cls._register_attribute(pl_url)
 172         cls._register_attribute(pl_ptn)
 173         cls._register_attribute(pl_user)
 174         cls._register_attribute(pl_password)
 175         #cls._register_attribute(site)
 176         cls._register_attribute(city)
 177         cls._register_attribute(country)
 178         cls._register_attribute(region)
 179         cls._register_attribute(architecture)
 180         cls._register_attribute(operating_system)
 181         cls._register_attribute(min_reliability)
 182         cls._register_attribute(max_reliability)
 183         cls._register_attribute(min_bandwidth)
 184         cls._register_attribute(max_bandwidth)
 185         cls._register_attribute(min_load)
 186         cls._register_attribute(max_load)
 187         cls._register_attribute(min_cpu)
 188         cls._register_attribute(max_cpu)
 189         cls._register_attribute(timeframe)
 190
 191
 192     def __init__(self, ec, guid):
 193         super(PlanetlabNode, self).__init__(ec, guid)
 194
 195         self._plapi = None
 196         self._node_to_provision = None
 197
 198     @property
 199     def plapi(self):
 200         if not self._plapi:
 201             pl_user = self.get("pluser")
 202             pl_pass = self.get("plpassword")
 203             pl_url = self.get("plcApiUrl")
 204             pl_ptn = self.get("plcApiPattern")
 205
 206             self._plapi =  PLCAPIFactory.get_api(pl_user, pl_pass, pl_url,
 207                     pl_ptn)
 208
 209         return self._plapi
 210
 211     def discover(self):
 212         """
 213         Based on the attributes defined by the user, discover the suitable nodes
 214         """
 215         hostname = self._get_hostname()
 216         print self.guid, hostname
 217         if hostname:
 218             # the user specified one particular node to be provisioned
 219             # check with PLCAPI if it is alvive
 220             node_id = self._query_if_alive(hostname=hostname)
 221             node_id = node_id.pop()
 222             print self.guid, node_id
 223
 224             # check that the node is not blacklisted or being provisioned
 225             # by other RM
 226             with PlanetlabNode.lock:
 227                 plist = self.plapi.reserved()
 228                 blist = self.plapi.blacklisted()
 229                 print self.guid,plist
 230                 print self.guid,blist
 231                 if node_id not in blist and node_id not in plist:
 232
 233                     # check that is really alive, by performing ping
 234                     ping_ok = self._do_ping(node_id)
 235                     if not ping_ok:
 236                         self._blacklist_node(node_id)
 237                         self.fail_node_not_alive(hostname)
 238                     else:
 239                         self._put_node_in_provision(node_id)
 240                         self._node_to_provision = node_id
 241                         super(PlanetlabNode, self).discover()
 242
 243                 else:
 244                     self.fail_node_not_available(hostname)
 245
 246         else:
 247             # the user specifies constraints based on attributes, zero, one or
 248             # more nodes can match these constraints
 249             nodes = self._filter_based_on_attributes()
 250             nodes_alive = self._query_if_alive(nodes)
 251
 252             # nodes that are already part of user's slice have the priority to
 253             # provisioned
 254             nodes_inslice = self._check_if_in_slice(nodes_alive)
 255             nodes_not_inslice = list(set(nodes_alive) - set(nodes_inslice))
 256
 257             node_id = None
 258             if nodes_inslice:
 259                 node_id = self._choose_random_node(nodes_inslice)
 260
 261             if not node_id:
 262                 # Either there were no matching nodes in the user's slice, or
 263                 # the nodes in the slice  were blacklisted or being provisioned
 264                 # by other RM. Note nodes_not_inslice is never empty
 265                 node_id = self._choose_random_node(nodes_not_inslice)
 266
 267             if node_id:
 268                 self._node_to_provision = node_id
 269                 super(PlanetlabNode, self).discover()
 270             else:
 271                self.fail_not_enough_nodes()
 272
 273     def provision(self):
 274         """
 275         Add node to user's slice after verifing that the node is functioning
 276         correctly
 277         """
 278         provision_ok = False
 279         ssh_ok = False
 280         proc_ok = False
 281         timeout = 120
 282
 283         while not provision_ok:
 284             node = self._node_to_provision
 285             # Adding try catch to set hostname because sometimes MyPLC fails
 286             # when trying to retrive node's hostname
 287             try:
 288                 self._set_hostname_attr(node)
 289             except:
 290                 with PlanetlabNode.lock:
 291                     self._blacklist_node(node)
 292                 self.discover()
 293                 continue
 294
 295             self._add_node_to_slice(node)
 296
 297             # check ssh connection
 298             t = 0
 299             while t < timeout and not ssh_ok:
 300
 301                 cmd = 'echo \'GOOD NODE\''
 302                 ((out, err), proc) = self.execute(cmd)
 303                 if out.find("GOOD NODE") < 0:
 304                     t = t + 60
 305                     time.sleep(60)
 306                     continue
 307                 else:
 308                     ssh_ok = True
 309                     continue
 310
 311             if not ssh_ok:
 312                 # the timeout was reach without establishing ssh connection
 313                 # the node is blacklisted, deleted from the slice, and a new
 314                 # node to provision is discovered
 315                 with PlanetlabNode.lock:
 316                     self._blacklist_node(node)
 317                     self._delete_node_from_slice(node)
 318                 self.set('hostname', None)
 319                 self.discover()
 320                 continue
 321
 322             # check /proc directory is mounted (ssh_ok = True)
 323             else:
 324                 cmd = 'mount |grep proc'
 325                 ((out, err), proc) = self.execute(cmd)
 326                 if out.find("/proc type proc") < 0:
 327                     with PlanetlabNode.lock:
 328                         self._blacklist_node(node)
 329                         self._delete_node_from_slice(node)
 330                     self.set('hostname', None)
 331                     self.discover()
 332                     continue
 333
 334                 else:
 335                     provision_ok = True
 336                     # set IP attribute
 337                     ip = self._get_ip(node)
 338                     self.set("ip", ip)
 339
 340         super(PlanetlabNode, self).provision()
 341
 342     def _filter_based_on_attributes(self):
 343         """
 344         Retrive the list of nodes ids that match user's constraints
 345         """
 346         # Map user's defined attributes with tagnames of PlanetLab
 347         timeframe = self.get("timeframe")[0]
 348         attr_to_tags = {
 349             'city' : 'city',
 350             'country' : 'country',
 351             'region' : 'region',
 352             'architecture' : 'arch',
 353             'operatingSystem' : 'fcdistro',
 354             #'site' : 'pldistro',
 355             'minReliability' : 'reliability%s' % timeframe,
 356             'maxReliability' : 'reliability%s' % timeframe,
 357             'minBandwidth' : 'bw%s' % timeframe,
 358             'maxBandwidth' : 'bw%s' % timeframe,
 359             'minLoad' : 'load%s' % timeframe,
 360             'maxLoad' : 'load%s' % timeframe,
 361             'minCpu' : 'cpu%s' % timeframe,
 362             'maxCpu' : 'cpu%s' % timeframe,
 363         }
 364
 365         nodes_id = []
 366         filters = {}
 367
 368         for attr_name, attr_obj in self._attrs.iteritems():
 369             attr_value = self.get(attr_name)
 370
 371             if attr_value is not None and attr_obj.flags == 8 and \
 372                 attr_name != 'timeframe':
 373
 374                 attr_tag = attr_to_tags[attr_name]
 375                 filters['tagname'] = attr_tag
 376
 377                 # filter nodes by fixed constraints e.g. operating system
 378                 if not 'min' in attr_name and not 'max' in attr_name:
 379                     filters['value'] = attr_value
 380                     nodes_id = self._filter_by_fixed_attr(filters, nodes_id)
 381
 382                 # filter nodes by range constraints e.g. max bandwidth
 383                 elif ('min' or 'max') in attr_name:
 384                     nodes_id = self._filter_by_range_attr(attr_name, attr_value, filters, nodes_id)
 385
 386         if not filters:
 387             nodes = self.plapi.get_nodes()
 388             for node in nodes:
 389                 nodes_id.append(node['node_id'])
 390
 391         return nodes_id
 392
 393
 394     def _filter_by_fixed_attr(self, filters, nodes_id):
 395         """
 396         Query PLCAPI for nodes ids matching fixed attributes defined by the
 397         user
 398         """
 399         node_tags = self.plapi.get_node_tags(filters)
 400         if node_tags is not None:
 401
 402             if len(nodes_id) == 0:
 403                 # first attribute being matched
 404                 for node_tag in node_tags:
 405                     nodes_id.append(node_tag['node_id'])
 406             else:
 407                 # remove the nodes ids that don't match the new attribute
 408                 # that is being match
 409
 410                 nodes_id_tmp = []
 411                 for node_tag in node_tags:
 412                     if node_tag['node_id'] in nodes_id:
 413                         nodes_id_tmp.append(node_tag['node_id'])
 414
 415                 if len(nodes_id_tmp):
 416                     nodes_id = set(nodes_id) & set(nodes_id_tmp)
 417                 else:
 418                     # no node from before match the new constraint
 419                     self.fail_discovery()
 420         else:
 421             # no nodes match the filter applied
 422             self.fail_discovery()
 423
 424         return nodes_id
 425
 426     def _filter_by_range_attr(self, attr_name, attr_value, filters, nodes_id):
 427         """
 428         Query PLCAPI for nodes ids matching attributes defined in a certain
 429         range, by the user
 430         """
 431         node_tags = self.plapi.get_node_tags(filters)
 432         if node_tags is not None:
 433
 434             if len(nodes_id) == 0:
 435                 # first attribute being matched
 436                 for node_tag in node_tags:
 437
 438                    # check that matches the min or max restriction
 439                     if 'min' in attr_name and node_tag['value'] != 'n/a' and \
 440                         float(node_tag['value']) > attr_value:
 441                         nodes_id.append(node_tag['node_id'])
 442
 443                     elif 'max' in attr_name and node_tag['value'] != 'n/a' and \
 444                         float(node_tag['value']) < attr_value:
 445                         nodes_id.append(node_tag['node_id'])
 446             else:
 447
 448                 # remove the nodes ids that don't match the new attribute
 449                 # that is being match
 450                 nodes_id_tmp = []
 451                 for node_tag in node_tags:
 452
 453                     # check that matches the min or max restriction and was a
 454                     # matching previous filters
 455                     if 'min' in attr_name and node_tag['value'] != 'n/a' and \
 456                         float(node_tag['value']) > attr_value and \
 457                         node_tag['node_id'] in nodes_id:
 458                         nodes_id_tmp.append(node_tag['node_id'])
 459
 460                     elif 'max' in attr_name and node_tag['value'] != 'n/a' and \
 461                         float(node_tag['value']) < attr_value and \
 462                         node_tag['node_id'] in nodes_id:
 463                         nodes_id_tmp.append(node_tag['node_id'])
 464
 465                 if len(nodes_id_tmp):
 466                     nodes_id = set(nodes_id) & set(nodes_id_tmp)
 467                 else:
 468                     # no node from before match the new constraint
 469                     self.fail_discovery()
 470
 471         else: #TODO CHECK
 472             # no nodes match the filter applied
 473             self.fail_discovery()
 474
 475         return nodes_id
 476
 477     def _query_if_alive(self, nodes_id=None, hostname=None):
 478         """
 479         Query PLCAPI for nodes that register activity recently, using filters
 480         related to the state of the node, e.g. last time it was contacted
 481         """
 482         if nodes_id is None and hostname is None:
 483             msg = "Specify nodes_id or hostname"
 484             raise RuntimeError, msg
 485
 486         if nodes_id is not None and hostname is not None:
 487             msg = "Specify either nodes_id or hostname"
 488             raise RuntimeError, msg
 489
 490         # define PL filters to check the node is alive
 491         filters = dict()
 492         filters['run_level'] = 'boot'
 493         filters['boot_state'] = 'boot'
 494         filters['node_type'] = 'regular'
 495         #filters['>last_contact'] =  int(time.time()) - 2*3600
 496
 497         # adding node_id or hostname to the filters to check for the particular
 498         # node
 499         if nodes_id:
 500             filters['node_id'] = list(nodes_id)
 501             alive_nodes_id = self._get_nodes_id(filters)
 502         elif hostname:
 503             filters['hostname'] = hostname
 504             alive_nodes_id = self._get_nodes_id(filters)
 505
 506         if len(alive_nodes_id) == 0:
 507             self.fail_node_not_alive(self, hostname)
 508         else:
 509             nodes_id = list()
 510             for node_id in alive_nodes_id:
 511                 nid = node_id['node_id']
 512                 nodes_id.append(nid)
 513
 514             return nodes_id
 515
 516     def _choose_random_node(self, nodes):
 517         """
 518         From the possible nodes for provision, choose randomly to decrese the
 519         probability of different RMs choosing the same node for provision
 520         """
 521         size = len(nodes)
 522         while size:
 523             size = size - 1
 524             index = randint(0, size)
 525             node_id = nodes[index]
 526             nodes[index] = nodes[size]
 527
 528             # check the node is not blacklisted or being provision by other RM
 529             # and perform ping to check that is really alive
 530             with PlanetlabNode.lock:
 531
 532                 blist = self.plapi.blacklisted()
 533                 plist = self.plapi.reserved()
 534                 if node_id not in blist and node_id not in plist:
 535                     ping_ok = self._do_ping(node_id)
 536                     print " ### ping_ok #### %s guid %s" % (ping_ok, self.guid)
 537                     if not ping_ok:
 538                         self._blacklist_node(node_id)
 539                     else:
 540                         # discovered node for provision, added to provision list
 541                         self._put_node_in_provision(node_id)
 542                         print "node_id %s , guid %s" % (node_id, self.guid)
 543                         return node_id
 544
 545     def _get_nodes_id(self, filters):
 546         return self.plapi.get_nodes(filters, fields=['node_id'])
 547
 548     def _add_node_to_slice(self, node_id):
 549         self.info(" Selected node to provision ")
 550         slicename = self.get("username")
 551         with PlanetlabNode.lock:
 552             slice_nodes = self.plapi.get_slice_nodes(slicename)
 553             slice_nodes.append(node_id)
 554             self.plapi.add_slice_nodes(slicename, slice_nodes)
 555
 556     def _delete_node_from_slice(self, node):
 557         self.warn(" Deleting node from slice ")
 558         slicename = self.get("username")
 559         self.plapi.delete_slice_node(slicename, [node])
 560
 561     def _get_hostname(self):
 562         hostname = self.get("hostname")
 563         ip = self.get("ip")
 564         if hostname:
 565             return hostname
 566         elif ip:
 567             hostname = sshfuncs.gethostbyname(ip)
 568             return hostname
 569         else:
 570             return None
 571
 572     def _set_hostname_attr(self, node):
 573         """
 574         Query PLCAPI for the hostname of a certain node id and sets the
 575         attribute hostname, it will over write the previous value
 576         """
 577         hostname = self.plapi.get_nodes(node, ['hostname'])
 578         self.set("hostname", hostname[0]['hostname'])
 579
 580     def _check_if_in_slice(self, nodes_id):
 581         """
 582         Query PLCAPI to find out if any node id from nodes_id is in the user's
 583         slice
 584         """
 585         slicename = self.get("username")
 586         slice_nodes = self.plapi.get_slice_nodes(slicename)
 587         nodes_inslice = list(set(nodes_id) & set(slice_nodes))
 588         return nodes_inslice
 589
 590     def _do_ping(self, node_id):
 591         """
 592         Perform ping command on node's IP matching node id
 593         """
 594         ping_ok = False
 595         ip = self._get_ip(node_id)
 596         print "ip de do_ping %s, guid %s" % (ip, self.guid)
 597         if not ip: return ping_ok
 598
 599         command = "ping -c2 %s" % ip
 600
 601         (out, err) = lexec(command)
 602         print "out de do_ping %s, guid %s" % (out, self.guid)
 603         if not out.find("2 received") < 0:
 604             ping_ok = True
 605
 606         print "ping_ok de do_ping %s, guid %s" % (ping_ok, self.guid)
 607         return ping_ok
 608
 609     def _blacklist_node(self, node):
 610         """
 611         Add node mal functioning node to blacklist
 612         """
 613         self.warn(" Blacklisting malfunctioning node ")
 614         self._plapi.blacklist_host(node)
 615
 616     def _put_node_in_provision(self, node):
 617         """
 618         Add node to the list of nodes being provisioned, in order for other RMs
 619         to not try to provision the same one again
 620         """
 621         self._plapi.reserve_host(node)
 622
 623     def _get_ip(self, node_id):
 624         """
 625         Query PLCAPI for the IP of a node with certain node id
 626         """
 627         hostname = self.plapi.get_nodes(node_id, ['hostname'])[0]
 628         print "#### HOSTNAME ##### %s ### guid %s " % (hostname['hostname'], self.guid)
 629         ip = sshfuncs.gethostbyname(hostname['hostname'])
 630         if not ip:
 631             # Fail while trying to find the IP
 632             return None
 633         return ip
 634
 635     def fail_discovery(self):
 636         self.fail()
 637         msg = "Discovery failed. No candidates found for node"
 638         self.error(msg)
 639         raise RuntimeError, msg
 640
 641     def fail_node_not_alive(self, hostname=None):
 642         self.fail()
 643         msg = "Node %s not alive" % hostname
 644         raise RuntimeError, msg
 645
 646     def fail_node_not_available(self, hostname):
 647         self.fail()
 648         msg = "Node %s not available for provisioning" % hostname
 649         raise RuntimeError, msg
 650
 651     def fail_not_enough_nodes(self):
 652         self.fail()
 653         msg = "Not enough nodes available for provisioning"
 654         raise RuntimeError, msg
 655
 656     def valid_connection(self, guid):
 657         # TODO: Validate!
 658         return True
 659
 660