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