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
  13 from nepi.util import server
  14
  15 class Node(object):
  16     BASEFILTERS = {
  17         # Map Node attribute to plcapi filter name
  18         'hostname' : 'hostname',
  19     }
  20
  21     TAGFILTERS = {
  22         # Map Node attribute to (<tag name>, <plcapi filter expression>)
  23         #   There are replacements that are applied with string formatting,
  24         #   so '%' has to be escaped as '%%'.
  25         'architecture' : ('arch','value'),
  26         'operating_system' : ('fcdistro','value'),
  27         'pl_distro' : ('pldistro','value'),
  28         'min_reliability' : ('reliability%(timeframe)s', ']value'),
  29         'max_reliability' : ('reliability%(timeframe)s', '[value'),
  30         'min_bandwidth' : ('bw%(timeframe)s', ']value'),
  31         'max_bandwidth' : ('bw%(timeframe)s', '[value'),
  32     }
  33
  34     DEPENDS_PIDFILE = '/tmp/nepi-depends.pid'
  35     DEPENDS_LOGFILE = '/tmp/nepi-depends.log'
  36
  37     def __init__(self, api=None):
  38         if not api:
  39             api = plcapi.PLCAPI()
  40         self._api = api
  41
  42         # Attributes
  43         self.hostname = None
  44         self.architecture = None
  45         self.operating_system = None
  46         self.pl_distro = None
  47         self.site = None
  48         self.emulation = None
  49         self.min_reliability = None
  50         self.max_reliability = None
  51         self.min_bandwidth = None
  52         self.max_bandwidth = None
  53         self.min_num_external_ifaces = None
  54         self.max_num_external_ifaces = None
  55         self.timeframe = 'm'
  56
  57         # Applications and routes add requirements to connected nodes
  58         self.required_packages = set()
  59         self.required_vsys = set()
  60         self.pythonpath = []
  61         self.env = collections.defaultdict(list)
  62
  63         # Testbed-derived attributes
  64         self.slicename = None
  65         self.ident_path = None
  66         self.server_key = None
  67         self.home_path = None
  68
  69         # Those are filled when an actual node is allocated
  70         self._node_id = None
  71
  72     @property
  73     def _nepi_testbed_environment_setup(self):
  74         command = cStringIO.StringIO()
  75         command.write('PYTHONPATH=$PYTHONPATH:%s' % (
  76             ':'.join(["${HOME}/"+server.shell_escape(s) for s in self.pythonpath])
  77         ))
  78         command.write(' PATH=$PATH:%s' % (
  79             ':'.join(["${HOME}/"+server.shell_escape(s) for s in self.pythonpath])
  80         ))
  81         if self.node.env:
  82             for envkey, envvals in self.node.env.iteritems():
  83                 for envval in envvals:
  84                     command.write(' %s=%s' % (envkey, envval))
  85         command.write(self.command)
  86         return command.getvalue()
  87
  88     def build_filters(self, target_filters, filter_map):
  89         for attr, tag in filter_map.iteritems():
  90             value = getattr(self, attr, None)
  91             if value is not None:
  92                 target_filters[tag] = value
  93         return target_filters
  94
  95     @property
  96     def applicable_filters(self):
  97         has = lambda att : getattr(self,att,None) is not None
  98         return (
  99             filter(has, self.BASEFILTERS.iterkeys())
 100             + filter(has, self.TAGFILTERS.iterkeys())
 101         )
 102
 103     def find_candidates(self, filter_slice_id=None):
 104         fields = ('node_id',)
 105         replacements = {'timeframe':self.timeframe}
 106
 107         # get initial candidates (no tag filters)
 108         basefilters = self.build_filters({}, self.BASEFILTERS)
 109         if filter_slice_id:
 110             basefilters['|slice_ids'] = (filter_slice_id,)
 111
 112         # keyword-only "pseudofilters"
 113         extra = {}
 114         if self.site:
 115             extra['peer'] = self.site
 116
 117         candidates = set(map(operator.itemgetter('node_id'),
 118             self._api.GetNodes(filters=basefilters, fields=fields, **extra)))
 119
 120         # filter by tag, one tag at a time
 121         applicable = self.applicable_filters
 122         for tagfilter in self.TAGFILTERS.iteritems():
 123             attr, (tagname, expr) = tagfilter
 124
 125             # don't bother if there's no filter defined
 126             if attr in applicable:
 127                 tagfilter = basefilters.copy()
 128                 tagfilter['tagname'] = tagname % replacements
 129                 tagfilter[expr % replacements] = getattr(self,attr)
 130                 tagfilter['node_id'] = list(candidates)
 131
 132                 candidates &= set(map(operator.itemgetter('node_id'),
 133                     self._api.GetNodeTags(filters=tagfilter, fields=fields)))
 134
 135         # filter by vsys tags - special case since it doesn't follow
 136         # the usual semantics
 137         if self.required_vsys:
 138             newcandidates = collections.defaultdict(set)
 139
 140             vsys_tags = self._api.GetNodeTags(
 141                 tagname='vsys',
 142                 node_id = list(candidates),
 143                 fields = ['node_id','value'])
 144
 145             vsys_tags = map(
 146                 operator.itemgetter(['node_id','value']),
 147                 vsys_tags)
 148
 149             required_vsys = self.required_vsys
 150             for node_id, value in vsys_tags:
 151                 if value in required_vsys:
 152                     newcandidates[value].add(node_id)
 153
 154             # take only those that have all the required vsys tags
 155             newcandidates = reduce(
 156                 lambda accum, new : accum & new,
 157                 newcandidates.itervalues(),
 158                 candidates)
 159
 160         # filter by iface count
 161         if self.min_num_external_ifaces is not None or self.max_num_external_ifaces is not None:
 162             # fetch interfaces for all, in one go
 163             filters = basefilters.copy()
 164             filters['node_id'] = list(candidates)
 165             ifaces = dict(map(operator.itemgetter('node_id','interface_ids'),
 166                 self._api.GetNodes(filters=basefilters, fields=('node_id','interface_ids')) ))
 167
 168             # filter candidates by interface count
 169             if self.min_num_external_ifaces is not None and self.max_num_external_ifaces is not None:
 170                 predicate = ( lambda node_id :
 171                     self.min_num_external_ifaces <= len(ifaces.get(node_id,())) <= self.max_num_external_ifaces )
 172             elif self.min_num_external_ifaces is not None:
 173                 predicate = ( lambda node_id :
 174                     self.min_num_external_ifaces <= len(ifaces.get(node_id,())) )
 175             else:
 176                 predicate = ( lambda node_id :
 177                     len(ifaces.get(node_id,())) <= self.max_num_external_ifaces )
 178
 179             candidates = set(filter(predicate, candidates))
 180
 181         return candidates
 182
 183     def assign_node_id(self, node_id):
 184         self._node_id = node_id
 185         self.fetch_node_info()
 186
 187     def fetch_node_info(self):
 188         info = self._api.GetNodes(self._node_id)[0]
 189         tags = dict( (t['tagname'],t['value'])
 190                      for t in self._api.GetNodeTags(node_id=self._node_id, fields=('tagname','value')) )
 191
 192         self.min_num_external_ifaces = None
 193         self.max_num_external_ifaces = None
 194         self.timeframe = 'm'
 195
 196         replacements = {'timeframe':self.timeframe}
 197         for attr, tag in self.BASEFILTERS.iteritems():
 198             if tag in info:
 199                 value = info[tag]
 200                 setattr(self, attr, value)
 201         for attr, (tag,_) in self.TAGFILTERS.iteritems():
 202             tag = tag % replacements
 203             if tag in tags:
 204                 value = tags[tag]
 205                 setattr(self, attr, value)
 206
 207         if 'peer_id' in info:
 208             self.site = self._api.peer_map[info['peer_id']]
 209
 210         if 'interface_ids' in info:
 211             self.min_num_external_ifaces = \
 212             self.max_num_external_ifaces = len(info['interface_ids'])
 213
 214         if 'ssh_rsa_key' in info:
 215             self.server_key = info['ssh_rsa_key']
 216
 217     def validate(self):
 218         if self.home_path is None:
 219             raise AssertionError, "Misconfigured node: missing home path"
 220         if self.ident_path is None or not os.access(self.ident_path, os.R_OK):
 221             raise AssertionError, "Misconfigured node: missing slice SSH key"
 222         if self.slicename is None:
 223             raise AssertionError, "Misconfigured node: unspecified slice"
 224
 225     def install_dependencies(self):
 226         if self.required_packages:
 227             # TODO: make dependant on the experiment somehow...
 228             pidfile = self.DEPENDS_PIDFILE
 229             logfile = self.DEPENDS_LOGFILE
 230
 231             # Start process in a "daemonized" way, using nohup and heavy
 232             # stdin/out redirection to avoid connection issues
 233             (out,err),proc = rspawn.remote_spawn(
 234                 "yum -y install %(packages)s" % {
 235                     'packages' : ' '.join(self.required_packages),
 236                 },
 237                 pidfile = pidfile,
 238                 stdout = logfile,
 239                 stderr = rspawn.STDOUT,
 240
 241                 host = self.hostname,
 242                 port = None,
 243                 user = self.slicename,
 244                 agent = None,
 245                 ident_key = self.ident_path,
 246                 server_key = self.server_key,
 247                 sudo = True
 248                 )
 249
 250             if proc.wait():
 251                 raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
 252
 253     def wait_dependencies(self, pidprobe=1, probe=0.5, pidmax=10, probemax=10):
 254         if self.required_packages:
 255             pidfile = self.DEPENDS_PIDFILE
 256
 257             # get PID
 258             pid = ppid = None
 259             for probenum in xrange(pidmax):
 260                 pidtuple = rspawn.remote_check_pid(
 261                     pidfile = pidfile,
 262                     host = self.hostname,
 263                     port = None,
 264                     user = self.slicename,
 265                     agent = None,
 266                     ident_key = self.ident_path,
 267                     server_key = self.server_key
 268                     )
 269                 if pidtuple:
 270                     pid, ppid = pidtuple
 271                     break
 272                 else:
 273                     time.sleep(pidprobe)
 274             else:
 275                 raise RuntimeError, "Failed to obtain pidfile for dependency installer"
 276
 277             # wait for it to finish
 278             while rspawn.RUNNING is rspawn.remote_status(
 279                     pid, ppid,
 280                     host = self.hostname,
 281                     port = None,
 282                     user = self.slicename,
 283                     agent = None,
 284                     ident_key = self.ident_path,
 285                     server_key = self.server_key
 286                     ):
 287                 time.sleep(probe)
 288                 probe = min(probemax, 1.5*probe)
 289
 290     def is_alive(self):
 291         # Make sure all the paths are created where
 292         # they have to be created for deployment
 293         (out,err),proc = server.popen_ssh_command(
 294             "echo 'ALIVE'",
 295             host = self.hostname,
 296             port = None,
 297             user = self.slicename,
 298             agent = None,
 299             ident_key = self.ident_path,
 300             server_key = self.server_key
 301             )
 302
 303         if proc.wait():
 304             return False
 305         elif not err and out.strip() == 'ALIVE':
 306             return True
 307         else:
 308             return False
 309
 310