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