src/nepi/testbeds/planetlab/node.py

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