src/nepi/testbeds/planetlab/node.py

   1 # -*- coding: utf-8 -*-
   2
   3 from constants import TESTBED_ID
   4 import plcapi
   5 import operator
   6 import rspawn
   7 import time
   8 import os
   9 import collections
  10 import cStringIO
  11 import resourcealloc
  12 import socket
  13 import sys
  14 import logging
  15 import ipaddr
  16 import operator
  17 import re
  18
  19 from nepi.util import server
  20 from nepi.util import parallel
  21
  22 import application
  23
  24 MAX_VROUTE_ROUTES = 5
  25
  26 class UnresponsiveNodeError(RuntimeError):
  27     pass
  28
  29 def _castproperty(typ, propattr):
  30     def _get(self):
  31         return getattr(self, propattr)
  32     def _set(self, value):
  33         if value is not None or (isinstance(value, basestring) and not value):
  34             value = typ(value)
  35         return setattr(self, propattr, value)
  36     def _del(self, value):
  37         return delattr(self, propattr)
  38     _get.__name__ = propattr + '_get'
  39     _set.__name__ = propattr + '_set'
  40     _del.__name__ = propattr + '_del'
  41     return property(_get, _set, _del)
  42
  43 class Node(object):
  44     BASEFILTERS = {
  45         # Map Node attribute to plcapi filter name
  46         'hostname' : 'hostname',
  47     }
  48
  49     TAGFILTERS = {
  50         # Map Node attribute to (<tag name>, <plcapi filter expression>)
  51         #   There are replacements that are applied with string formatting,
  52         #   so '%' has to be escaped as '%%'.
  53         'architecture' : ('arch','value'),
  54         'operatingSystem' : ('fcdistro','value'),
  55         'pl_distro' : ('pldistro','value'),
  56         'city' : ('city','value'),
  57         'country' : ('country','value'),
  58         'region' : ('region','value'),
  59         'minReliability' : ('reliability%(timeframe)s', ']value'),
  60         'maxReliability' : ('reliability%(timeframe)s', '[value'),
  61         'minBandwidth' : ('bw%(timeframe)s', ']value'),
  62         'maxBandwidth' : ('bw%(timeframe)s', '[value'),
  63         'minLoad' : ('load%(timeframe)s', ']value'),
  64         'maxLoad' : ('load%(timeframe)s', '[value'),
  65         'minCpu' : ('cpu%(timeframe)s', ']value'),
  66         'maxCpu' : ('cpu%(timeframe)s', '[value'),
  67     }
  68
  69     RATE_FACTORS = (
  70         # (<tag name>, <weight>, <default>)
  71         ('bw%(timeframe)s', -0.001, 1024.0),
  72         ('cpu%(timeframe)s', 0.1, 40.0),
  73         ('load%(timeframe)s', -0.2, 3.0),
  74         ('reliability%(timeframe)s', 1, 100.0),
  75     )
  76
  77     DEPENDS_PIDFILE = '/tmp/nepi-depends.pid'
  78     DEPENDS_LOGFILE = '/tmp/nepi-depends.log'
  79     RPM_FUSION_URL = 'http://download1.rpmfusion.org/free/fedora/rpmfusion-free-release-stable.noarch.rpm'
  80     RPM_FUSION_URL_F12 = 'http://download1.rpmfusion.org/free/fedora/releases/12/Everything/x86_64/os/rpmfusion-free-release-12-1.noarch.rpm'
  81
  82     minReliability = _castproperty(float, '_minReliability')
  83     maxReliability = _castproperty(float, '_maxReliability')
  84     minBandwidth = _castproperty(float, '_minBandwidth')
  85     maxBandwidth = _castproperty(float, '_maxBandwidth')
  86     minCpu = _castproperty(float, '_minCpu')
  87     maxCpu = _castproperty(float, '_maxCpu')
  88     minLoad = _castproperty(float, '_minLoad')
  89     maxLoad = _castproperty(float, '_maxLoad')
  90
  91     def __init__(self, api=None, sliceapi=None):
  92         if not api:
  93             api = plcapi.PLCAPI()
  94         self._api = api
  95         self._sliceapi = sliceapi
  96
  97         # Attributes
  98         self.hostname = None
  99         self.architecture = None
 100         self.operatingSystem = None
 101         self.pl_distro = None
 102         self.site = None
 103         self.city = None
 104         self.country = None
 105         self.region = None
 106         self.minReliability = None
 107         self.maxReliability = None
 108         self.minBandwidth = None
 109         self.maxBandwidth = None
 110         self.minCpu = None
 111         self.maxCpu = None
 112         self.minLoad = None
 113         self.maxLoad = None
 114         self.min_num_external_ifaces = None
 115         self.max_num_external_ifaces = None
 116         self.timeframe = 'm'
 117
 118         # Applications and routes add requirements to connected nodes
 119         self.required_packages = set()
 120         self.required_vsys = set()
 121         self.pythonpath = []
 122         self.rpmFusion = False
 123         self.env = collections.defaultdict(list)
 124
 125         # Some special applications - initialized when connected
 126         self.multicast_forwarder = None
 127
 128         # Testbed-derived attributes
 129         self.slicename = None
 130         self.ident_path = None
 131         self.server_key = None
 132         self.home_path = None
 133         self.enable_cleanup = False
 134
 135         # Those are filled when an actual node is allocated
 136         self._node_id = None
 137         self._yum_dependencies = None
 138         self._installed = False
 139
 140         # Logging
 141         self._logger = logging.getLogger('nepi.testbeds.planetlab')
 142
 143     def _nepi_testbed_environment_setup_get(self):
 144         command = cStringIO.StringIO()
 145         command.write('export PYTHONPATH=$PYTHONPATH:%s' % (
 146             ':'.join(["${HOME}/"+server.shell_escape(s) for s in self.pythonpath])
 147         ))
 148         command.write(' ; export PATH=$PATH:%s' % (
 149             ':'.join(["${HOME}/"+server.shell_escape(s) for s in self.pythonpath])
 150         ))
 151         if self.env:
 152             for envkey, envvals in self.env.iteritems():
 153                 for envval in envvals:
 154                     command.write(' ; export %s=%s' % (envkey, envval))
 155         return command.getvalue()
 156     def _nepi_testbed_environment_setup_set(self, value):
 157         pass
 158     _nepi_testbed_environment_setup = property(
 159         _nepi_testbed_environment_setup_get,
 160         _nepi_testbed_environment_setup_set)
 161
 162     def build_filters(self, target_filters, filter_map):
 163         for attr, tag in filter_map.iteritems():
 164             value = getattr(self, attr, None)
 165             if value is not None:
 166                 target_filters[tag] = value
 167         return target_filters
 168
 169     @property
 170     def applicable_filters(self):
 171         has = lambda att : getattr(self,att,None) is not None
 172         return (
 173             filter(has, self.BASEFILTERS.iterkeys())
 174             + filter(has, self.TAGFILTERS.iterkeys())
 175         )
 176
 177     def find_candidates(self, filter_slice_id=None):
 178         self._logger.info("Finding candidates for %s", self.make_filter_description())
 179
 180         fields = ('node_id',)
 181         replacements = {'timeframe':self.timeframe}
 182
 183         # get initial candidates (no tag filters)
 184         basefilters = self.build_filters({}, self.BASEFILTERS)
 185         rootfilters = basefilters.copy()
 186         if filter_slice_id:
 187             basefilters['|slice_ids'] = (filter_slice_id,)
 188
 189         # only pick healthy nodes
 190         basefilters['run_level'] = 'boot'
 191         basefilters['boot_state'] = 'boot'
 192         basefilters['node_type'] = 'regular' # nepi can only handle regular nodes (for now)
 193         basefilters['>last_contact'] = int(time.time()) - 5*3600 # allow 5h out of contact, for timezone discrepancies
 194
 195         # keyword-only "pseudofilters"
 196         extra = {}
 197         if self.site:
 198             extra['peer'] = self.site
 199
 200         candidates = set(map(operator.itemgetter('node_id'),
 201             self._sliceapi.GetNodes(filters=basefilters, fields=fields, **extra)))
 202
 203         # filter by tag, one tag at a time
 204         applicable = self.applicable_filters
 205         for tagfilter in self.TAGFILTERS.iteritems():
 206             attr, (tagname, expr) = tagfilter
 207
 208             # don't bother if there's no filter defined
 209             if attr in applicable:
 210                 tagfilter = rootfilters.copy()
 211                 tagfilter['tagname'] = tagname % replacements
 212                 tagfilter[expr % replacements] = getattr(self,attr)
 213                 tagfilter['node_id'] = list(candidates)
 214
 215                 candidates &= set(map(operator.itemgetter('node_id'),
 216                     self._sliceapi.GetNodeTags(filters=tagfilter, fields=fields)))
 217
 218         # filter by vsys tags - special case since it doesn't follow
 219         # the usual semantics
 220         if self.required_vsys:
 221             newcandidates = collections.defaultdict(set)
 222
 223             vsys_tags = self._sliceapi.GetNodeTags(
 224                 tagname='vsys',
 225                 node_id = list(candidates),
 226                 fields = ['node_id','value'])
 227
 228             vsys_tags = map(
 229                 operator.itemgetter(['node_id','value']),
 230                 vsys_tags)
 231
 232             required_vsys = self.required_vsys
 233             for node_id, value in vsys_tags:
 234                 if value in required_vsys:
 235                     newcandidates[value].add(node_id)
 236
 237             # take only those that have all the required vsys tags
 238             newcandidates = reduce(
 239                 lambda accum, new : accum & new,
 240                 newcandidates.itervalues(),
 241                 candidates)
 242
 243         # filter by iface count
 244         if self.min_num_external_ifaces is not None or self.max_num_external_ifaces is not None:
 245             # fetch interfaces for all, in one go
 246             filters = basefilters.copy()
 247             filters['node_id'] = list(candidates)
 248             ifaces = dict(map(operator.itemgetter('node_id','interface_ids'),
 249                 self._sliceapi.GetNodes(filters=basefilters, fields=('node_id','interface_ids')) ))
 250
 251             # filter candidates by interface count
 252             if self.min_num_external_ifaces is not None and self.max_num_external_ifaces is not None:
 253                 predicate = ( lambda node_id :
 254                     self.min_num_external_ifaces <= len(ifaces.get(node_id,())) <= self.max_num_external_ifaces )
 255             elif self.min_num_external_ifaces is not None:
 256                 predicate = ( lambda node_id :
 257                     self.min_num_external_ifaces <= len(ifaces.get(node_id,())) )
 258             else:
 259                 predicate = ( lambda node_id :
 260                     len(ifaces.get(node_id,())) <= self.max_num_external_ifaces )
 261
 262             candidates = set(filter(predicate, candidates))
 263
 264         # make sure hostnames are resolvable
 265         hostnames = None
 266         if candidates:
 267             self._logger.info("  Found %s candidates. Checking for reachability...", len(candidates))
 268
 269             hostnames = dict(map(operator.itemgetter('node_id','hostname'),
 270                 self._api.GetNodes(list(candidates), ['node_id','hostname'])
 271             ))
 272             def resolvable(node_id):
 273                 try:
 274                     addr = socket.gethostbyname(hostnames[node_id])
 275                     return addr is not None
 276                 except:
 277                     return False
 278             candidates = set(parallel.pfilter(resolvable, candidates,
 279                 maxthreads = 16))
 280
 281             self._logger.info("  Found %s reachable candidates.", len(candidates))
 282
 283             for h in hostnames.keys():
 284                 if h not in candidates:
 285                     del hostnames[h]
 286
 287             hostnames = dict((v,k) for k, v in hostnames.iteritems())
 288
 289         return hostnames
 290
 291     def make_filter_description(self):
 292         """
 293         Makes a human-readable description of filtering conditions
 294         for find_candidates.
 295         """
 296
 297         # get initial candidates (no tag filters)
 298         filters = self.build_filters({}, self.BASEFILTERS)
 299
 300         # keyword-only "pseudofilters"
 301         if self.site:
 302             filters['peer'] = self.site
 303
 304         # filter by tag, one tag at a time
 305         applicable = self.applicable_filters
 306         for tagfilter in self.TAGFILTERS.iteritems():
 307             attr, (tagname, expr) = tagfilter
 308
 309             # don't bother if there's no filter defined
 310             if attr in applicable:
 311                 filters[attr] = getattr(self,attr)
 312
 313         # filter by vsys tags - special case since it doesn't follow
 314         # the usual semantics
 315         if self.required_vsys:
 316             filters['vsys'] = ','.join(list(self.required_vsys))
 317
 318         # filter by iface count
 319         if self.min_num_external_ifaces is not None or self.max_num_external_ifaces is not None:
 320             filters['num_ifaces'] = '-'.join([
 321                 str(self.min_num_external_ifaces or '0'),
 322                 str(self.max_num_external_ifaces or 'inf')
 323             ])
 324
 325         return '; '.join(map('%s: %s'.__mod__,filters.iteritems()))
 326
 327     def assign_node_id(self, node_id):
 328         self._node_id = node_id
 329         self.fetch_node_info()
 330
 331     def unassign_node(self):
 332         self._node_id = None
 333         self.hostip = None
 334
 335         try:
 336             orig_attrs = self.__orig_attrs
 337         except AttributeError:
 338             return
 339
 340         for key, value in orig_attrs.iteritems():
 341             setattr(self, key, value)
 342         del self.__orig_attrs
 343
 344     def rate_nodes(self, nodes):
 345         rates = collections.defaultdict(int)
 346         tags = collections.defaultdict(dict)
 347         replacements = {'timeframe':self.timeframe}
 348         tagnames = [ tagname % replacements
 349                      for tagname, weight, default in self.RATE_FACTORS ]
 350
 351         taginfo = self._sliceapi.GetNodeTags(
 352             node_id=list(nodes),
 353             tagname=tagnames,
 354             fields=('node_id','tagname','value'))
 355
 356         unpack = operator.itemgetter('node_id','tagname','value')
 357         for value in taginfo:
 358             node, tagname, value = unpack(value)
 359             if value and value.lower() != 'n/a':
 360                 tags[tagname][int(node)] = float(value)
 361
 362         for tagname, weight, default in self.RATE_FACTORS:
 363             taginfo = tags[tagname % replacements].get
 364             for node in nodes:
 365                 rates[node] += weight * taginfo(node,default)
 366
 367         return map(rates.__getitem__, nodes)
 368
 369     def fetch_node_info(self):
 370         orig_attrs = {}
 371
 372         info, tags = self._sliceapi.GetNodeInfo(self._node_id)
 373         info = info[0]
 374
 375         tags = dict( (t['tagname'],t['value'])
 376                      for t in tags )
 377
 378         orig_attrs['min_num_external_ifaces'] = self.min_num_external_ifaces
 379         orig_attrs['max_num_external_ifaces'] = self.max_num_external_ifaces
 380         self.min_num_external_ifaces = None
 381         self.max_num_external_ifaces = None
 382         self.timeframe = 'm'
 383
 384         replacements = {'timeframe':self.timeframe}
 385         for attr, tag in self.BASEFILTERS.iteritems():
 386             if tag in info:
 387                 value = info[tag]
 388                 if hasattr(self, attr):
 389                     orig_attrs[attr] = getattr(self, attr)
 390                 setattr(self, attr, value)
 391         for attr, (tag,_) in self.TAGFILTERS.iteritems():
 392             tag = tag % replacements
 393             if tag in tags:
 394                 value = tags[tag]
 395                 if hasattr(self, attr):
 396                     orig_attrs[attr] = getattr(self, attr)
 397                 if not value or value.lower() == 'n/a':
 398                     value = None
 399                 setattr(self, attr, value)
 400
 401         if 'peer_id' in info:
 402             orig_attrs['site'] = self.site
 403             self.site = self._api.peer_map[info['peer_id']]
 404
 405         if 'interface_ids' in info:
 406             self.min_num_external_ifaces = \
 407             self.max_num_external_ifaces = len(info['interface_ids'])
 408
 409         if 'ssh_rsa_key' in info:
 410             orig_attrs['server_key'] = self.server_key
 411             self.server_key = info['ssh_rsa_key']
 412
 413         self.hostip = socket.gethostbyname(self.hostname)
 414
 415         try:
 416             self.__orig_attrs
 417         except AttributeError:
 418             self.__orig_attrs = orig_attrs
 419
 420     def validate(self):
 421         if self.home_path is None:
 422             raise AssertionError, "Misconfigured node: missing home path"
 423         if self.ident_path is None or not os.access(self.ident_path, os.R_OK):
 424             raise AssertionError, "Misconfigured node: missing slice SSH key"
 425         if self.slicename is None:
 426             raise AssertionError, "Misconfigured node: unspecified slice"
 427
 428     def recover(self):
 429         # Mark dependencies installed
 430         self._installed = True
 431
 432         # Clear load attributes, they impair re-discovery
 433         self.minReliability = \
 434         self.maxReliability = \
 435         self.minBandwidth = \
 436         self.maxBandwidth = \
 437         self.minCpu = \
 438         self.maxCpu = \
 439         self.minLoad = \
 440         self.maxLoad = None
 441
 442     def install_dependencies(self):
 443         if self.required_packages and not self._installed:
 444             # If we need rpmfusion, we must install the repo definition and the gpg keys
 445             if self.rpmFusion:
 446                 if self.operatingSystem == 'f12':
 447                     # Fedora 12 requires a different rpmfusion package
 448                     RPM_FUSION_URL = self.RPM_FUSION_URL_F12
 449                 else:
 450                     # This one works for f13+
 451                     RPM_FUSION_URL = self.RPM_FUSION_URL
 452
 453                 rpmFusion = (
 454                   'rpm -q $(rpm -q -p %(RPM_FUSION_URL)s) || sudo -S rpm -i %(RPM_FUSION_URL)s'
 455                 ) % {
 456                     'RPM_FUSION_URL' : RPM_FUSION_URL
 457                 }
 458             else:
 459                 rpmFusion = ''
 460
 461             if rpmFusion:
 462                 (out,err),proc = server.popen_ssh_command(
 463                     rpmFusion,
 464                     host = self.hostname,
 465                     port = None,
 466                     user = self.slicename,
 467                     agent = None,
 468                     ident_key = self.ident_path,
 469                     server_key = self.server_key,
 470                     timeout = 600,
 471                     )
 472
 473                 if proc.wait():
 474                     if self.check_bad_host(out,err):
 475                         self.blacklist()
 476                     raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
 477
 478             # Launch p2p yum dependency installer
 479             self._yum_dependencies.async_setup()
 480
 481     def wait_provisioning(self, timeout = 20*60):
 482         # Wait for the p2p installer
 483         sleeptime = 1.0
 484         totaltime = 0.0
 485         while not self.is_alive():
 486             time.sleep(sleeptime)
 487             totaltime += sleeptime
 488             sleeptime = min(30.0, sleeptime*1.5)
 489
 490             if totaltime > timeout:
 491                 # PlanetLab has a 15' delay on configuration propagation
 492                 # If we're above that delay, the unresponsiveness is not due
 493                 # to this delay.
 494                 if not self.is_alive(verbose=True):
 495                     raise UnresponsiveNodeError, "Unresponsive host %s" % (self.hostname,)
 496
 497         # Ensure the node is clean (no apps running that could interfere with operations)
 498         if self.enable_cleanup:
 499             self.do_cleanup()
 500
 501     def wait_dependencies(self, pidprobe=1, probe=0.5, pidmax=10, probemax=10):
 502         # Wait for the p2p installer
 503         if self._yum_dependencies and not self._installed:
 504             self._yum_dependencies.async_setup_wait()
 505             self._installed = True
 506
 507     def is_alive(self, verbose = False):
 508         # Make sure all the paths are created where
 509         # they have to be created for deployment
 510         (out,err),proc = server.eintr_retry(server.popen_ssh_command)(
 511             "echo 'ALIVE'",
 512             host = self.hostname,
 513             port = None,
 514             user = self.slicename,
 515             agent = None,
 516             ident_key = self.ident_path,
 517             server_key = self.server_key,
 518             timeout = 60,
 519             err_on_timeout = False,
 520             persistent = False
 521             )
 522
 523         if proc.wait():
 524             if verbose:
 525                 self._logger.warn("Unresponsive node %s got:\n%s%s", self.hostname, out, err)
 526             return False
 527         elif not err and out.strip() == 'ALIVE':
 528             return True
 529         else:
 530             if verbose:
 531                 self._logger.warn("Unresponsive node %s got:\n%s%s", self.hostname, out, err)
 532             return False
 533
 534     def destroy(self):
 535         if self.enable_cleanup:
 536             self.do_cleanup()
 537
 538     def blacklist(self):
 539         if self._node_id:
 540             self._logger.warn("Blacklisting malfunctioning node %s", self.hostname)
 541             import util
 542             util.appendBlacklist(self.hostname)
 543
 544     def do_cleanup(self):
 545         if self.testbed().recovering:
 546             # WOW - not now
 547             return
 548
 549         self._logger.info("Cleaning up %s", self.hostname)
 550
 551         cmds = [
 552             "sudo -S killall python tcpdump || /bin/true ; "
 553             "sudo -S killall python tcpdump || /bin/true ; "
 554             "sudo -S kill $(ps -N -T -o pid --no-heading | grep -v $PPID | sort) || /bin/true ",
 555             "sudo -S killall -u %(slicename)s || /bin/true ",
 556             "sudo -S killall -u root || /bin/true ",
 557             "sudo -S killall -u %(slicename)s || /bin/true ",
 558             "sudo -S killall -u root || /bin/true ",
 559         ]
 560
 561         for cmd in cmds:
 562             (out,err),proc = server.popen_ssh_command(
 563                 # Some apps need two kills
 564                 cmd % {
 565                     'slicename' : self.slicename ,
 566                 },
 567                 host = self.hostname,
 568                 port = None,
 569                 user = self.slicename,
 570                 agent = None,
 571                 ident_key = self.ident_path,
 572                 server_key = self.server_key,
 573                 tty = True, # so that ps -N -T works as advertised...
 574                 timeout = 60,
 575                 retry = 3
 576                 )
 577             proc.wait()
 578
 579     def prepare_dependencies(self):
 580         # Configure p2p yum dependency installer
 581         if self.required_packages and not self._installed:
 582             self._yum_dependencies = application.YumDependency(self._api)
 583             self._yum_dependencies.node = self
 584             self._yum_dependencies.home_path = "nepi-yumdep"
 585             self._yum_dependencies.depends = ' '.join(self.required_packages)
 586
 587     def routing_method(self, routes, vsys_vnet):
 588         """
 589         There are two methods, vroute and sliceip.
 590
 591         vroute:
 592             Modifies the node's routing table directly, validating that the IP
 593             range lies within the network given by the slice's vsys_vnet tag.
 594             This method is the most scalable for very small routing tables
 595             that need not modify other routes (including the default)
 596
 597         sliceip:
 598             Uses policy routing and iptables filters to create per-sliver
 599             routing tables. It's the most flexible way, but it doesn't scale
 600             as well since only 155 routing tables can be created this way.
 601
 602         This method will return the most appropriate routing method, which will
 603         prefer vroute for small routing tables.
 604         """
 605
 606         # For now, sliceip results in kernel panics
 607         # so we HAVE to use vroute
 608         return 'vroute'
 609
 610         # We should not make the routing table grow too big
 611         if len(routes) > MAX_VROUTE_ROUTES:
 612             return 'sliceip'
 613
 614         vsys_vnet = ipaddr.IPNetwork(vsys_vnet)
 615         for route in routes:
 616             dest, prefix, nexthop, metric = route
 617             dest = ipaddr.IPNetwork("%s/%d" % (dest,prefix))
 618             nexthop = ipaddr.IPAddress(nexthop)
 619             if dest not in vsys_vnet or nexthop not in vsys_vnet:
 620                 return 'sliceip'
 621
 622         return 'vroute'
 623
 624     def format_route(self, route, dev, method, action):
 625         dest, prefix, nexthop, metric = route
 626         if method == 'vroute':
 627             return (
 628                 "%s %s%s gw %s %s" % (
 629                     action,
 630                     dest,
 631                     (("/%d" % (prefix,)) if prefix and prefix != 32 else ""),
 632                     nexthop,
 633                     dev,
 634                 )
 635             )
 636         elif method == 'sliceip':
 637             return (
 638                 "route %s to %s%s via %s metric %s dev %s" % (
 639                     action,
 640                     dest,
 641                     (("/%d" % (prefix,)) if prefix and prefix != 32 else ""),
 642                     nexthop,
 643                     metric or 1,
 644                     dev,
 645                 )
 646             )
 647         else:
 648             raise AssertionError, "Unknown method"
 649
 650     def _annotate_routes_with_devs(self, routes, devs, method):
 651         dev_routes = []
 652         for route in routes:
 653             for dev in devs:
 654                 if dev.routes_here(route):
 655                     dev_routes.append(tuple(route) + (dev.if_name,))
 656
 657                     # Stop checking
 658                     break
 659             else:
 660                 if method == 'sliceip':
 661                     dev_routes.append(tuple(route) + ('eth0',))
 662                 else:
 663                     raise RuntimeError, "Route %s cannot be bound to any virtual interface " \
 664                         "- PL can only handle rules over virtual interfaces. Candidates are: %s" % (route,devs)
 665         return dev_routes
 666
 667     def configure_routes(self, routes, devs, vsys_vnet):
 668         """
 669         Add the specified routes to the node's routing table
 670         """
 671         rules = []
 672         method = self.routing_method(routes, vsys_vnet)
 673         tdevs = set()
 674
 675         # annotate routes with devices
 676         dev_routes = self._annotate_routes_with_devs(routes, devs, method)
 677         for route in dev_routes:
 678             route, dev = route[:-1], route[-1]
 679
 680             # Schedule rule
 681             tdevs.add(dev)
 682             rules.append(self.format_route(route, dev, method, 'add'))
 683
 684         if method == 'sliceip':
 685             rules = map('enable '.__add__, tdevs) + rules
 686
 687         self._logger.info("Setting up routes for %s using %s", self.hostname, method)
 688         self._logger.debug("Routes for %s:\n\t%s", self.hostname, '\n\t'.join(rules))
 689
 690         self.apply_route_rules(rules, method)
 691
 692         self._configured_routes = set(routes)
 693         self._configured_devs = tdevs
 694         self._configured_method = method
 695
 696     def reconfigure_routes(self, routes, devs, vsys_vnet):
 697         """
 698         Updates the routes in the node's routing table to match
 699         the given route list
 700         """
 701         method = self._configured_method
 702
 703         dev_routes = self._annotate_routes_with_devs(routes, devs, method)
 704
 705         current = self._configured_routes
 706         current_devs = self._configured_devs
 707
 708         new = set(dev_routes)
 709         new_devs = set(map(operator.itemgetter(-1), dev_routes))
 710
 711         deletions = current - new
 712         insertions = new - current
 713
 714         dev_deletions = current_devs - new_devs
 715         dev_insertions = new_devs - current_devs
 716
 717         # Generate rules
 718         rules = []
 719
 720         # Rule deletions first
 721         for route in deletions:
 722             route, dev = route[:-1], route[-1]
 723             rules.append(self.format_route(route, dev, method, 'del'))
 724
 725         if method == 'sliceip':
 726             # Dev deletions now
 727             rules.extend(map('disable '.__add__, dev_deletions))
 728
 729             # Dev insertions now
 730             rules.extend(map('enable '.__add__, dev_insertions))
 731
 732         # Rule insertions now
 733         for route in insertions:
 734             route, dev = route[:-1], dev[-1]
 735             rules.append(self.format_route(route, dev, method, 'add'))
 736
 737         # Apply
 738         self.apply_route_rules(rules, method)
 739
 740         self._configured_routes = dev_routes
 741         self._configured_devs = new_devs
 742
 743     def apply_route_rules(self, rules, method):
 744         (out,err),proc = server.popen_ssh_command(
 745             "( sudo -S bash -c 'cat /vsys/%(method)s.out >&2' & ) ; sudo -S bash -c 'cat > /vsys/%(method)s.in' ; sleep 0.5" % dict(
 746                 home = server.shell_escape(self.home_path),
 747                 method = method),
 748             host = self.hostname,
 749             port = None,
 750             user = self.slicename,
 751             agent = None,
 752             ident_key = self.ident_path,
 753             server_key = self.server_key,
 754             stdin = '\n'.join(rules),
 755             timeout = 300
 756             )
 757
 758         if proc.wait() or err:
 759             raise RuntimeError, "Could not set routes (%s) errors: %s%s" % (rules,out,err)
 760         elif out or err:
 761             logger.debug("%s said: %s%s", method, out, err)
 762
 763     def check_bad_host(self, out, err):
 764         badre = re.compile(r'(?:'
 765                            r"curl: [(]\d+[)] Couldn't resolve host 'download1[.]rpmfusion[.]org'"
 766                            r'|Error: disk I/O error'
 767                            r')',
 768                            re.I)
 769         return badre.search(out) or badre.search(err)