sfa/planetlab/plaggregate.py

   1 #!/usr/bin/python
   2 from collections import defaultdict
   3 from sfa.util.xrn import Xrn, hrn_to_urn, urn_to_hrn
   4 from sfa.util.sfatime import utcparse, datetime_to_string
   5 from sfa.util.sfalogging import logger
   6 from sfa.util.faults import SliverDoesNotExist
   7 from sfa.rspecs.rspec import RSpec
   8 from sfa.rspecs.elements.hardware_type import HardwareType
   9 from sfa.rspecs.elements.node import Node
  10 from sfa.rspecs.elements.link import Link
  11 from sfa.rspecs.elements.sliver import Sliver
  12 from sfa.rspecs.elements.login import Login
  13 from sfa.rspecs.elements.location import Location
  14 from sfa.rspecs.elements.interface import Interface
  15 from sfa.rspecs.elements.services import Services
  16 from sfa.rspecs.elements.pltag import PLTag
  17 from sfa.rspecs.elements.lease import Lease
  18 from sfa.rspecs.elements.granularity import Granularity
  19 from sfa.rspecs.version_manager import VersionManager
  20
  21 from sfa.planetlab.plxrn import PlXrn, hostname_to_urn, hrn_to_pl_slicename, slicename_to_hrn
  22 from sfa.planetlab.vlink import get_tc_rate
  23 from sfa.planetlab.topology import Topology
  24 from sfa.storage.alchemy import dbsession
  25 from sfa.storage.model import SliverAllocation
  26
  27
  28 import time
  29
  30 class PlAggregate:
  31
  32     def __init__(self, driver):
  33         self.driver = driver
  34
  35     def get_nodes(self, options={}):
  36         filter = {'peer_id': None}
  37         geni_available = options.get('geni_available')
  38         if geni_available == True:
  39             filter['boot_state'] = 'boot'
  40         nodes = self.driver.shell.GetNodes(filter)
  41
  42         return nodes
  43
  44     def get_sites(self, filter={}):
  45         sites = {}
  46         for site in self.driver.shell.GetSites(filter):
  47             sites[site['site_id']] = site
  48         return sites
  49
  50     def get_interfaces(self, filter={}):
  51         interfaces = {}
  52         for interface in self.driver.shell.GetInterfaces(filter):
  53             iface = Interface()
  54             if interface['bwlimit']:
  55                 interface['bwlimit'] = str(int(interface['bwlimit'])/1000)
  56             interfaces[interface['interface_id']] = interface
  57         return interfaces
  58
  59     def get_links(self, sites, nodes, interfaces):
  60
  61         topology = Topology()
  62         links = []
  63         for (site_id1, site_id2) in topology:
  64             site_id1 = int(site_id1)
  65             site_id2 = int(site_id2)
  66             link = Link()
  67             if not site_id1 in sites or site_id2 not in sites:
  68                 continue
  69             site1 = sites[site_id1]
  70             site2 = sites[site_id2]
  71             # get hrns
  72             site1_hrn = self.driver.hrn + '.' + site1['login_base']
  73             site2_hrn = self.driver.hrn + '.' + site2['login_base']
  74
  75             for s1_node_id in site1['node_ids']:
  76                 for s2_node_id in site2['node_ids']:
  77                     if s1_node_id not in nodes or s2_node_id not in nodes:
  78                         continue
  79                     node1 = nodes[s1_node_id]
  80                     node2 = nodes[s2_node_id]
  81                     # set interfaces
  82                     # just get first interface of the first node
  83                     if1_xrn = PlXrn(auth=self.driver.hrn, interface='node%s:eth0' % (node1['node_id']))
  84                     if1_ipv4 = interfaces[node1['interface_ids'][0]]['ip']
  85                     if2_xrn = PlXrn(auth=self.driver.hrn, interface='node%s:eth0' % (node2['node_id']))
  86                     if2_ipv4 = interfaces[node2['interface_ids'][0]]['ip']
  87
  88                     if1 = Interface({'component_id': if1_xrn.urn, 'ipv4': if1_ipv4} )
  89                     if2 = Interface({'component_id': if2_xrn.urn, 'ipv4': if2_ipv4} )
  90
  91                     # set link
  92                     link = Link({'capacity': '1000000', 'latency': '0', 'packet_loss': '0', 'type': 'ipv4'})
  93                     link['interface1'] = if1
  94                     link['interface2'] = if2
  95                     link['component_name'] = "%s:%s" % (site1['login_base'], site2['login_base'])
  96                     link['component_id'] = PlXrn(auth=self.driver.hrn, interface=link['component_name']).get_urn()
  97                     link['component_manager_id'] =  hrn_to_urn(self.driver.hrn, 'authority+am')
  98                     links.append(link)
  99
 100         return links
 101
 102     def get_node_tags(self, filter={}):
 103         node_tags = {}
 104         for node_tag in self.driver.shell.GetNodeTags(filter):
 105             node_tags[node_tag['node_tag_id']] = node_tag
 106         return node_tags
 107
 108     def get_pl_initscripts(self, filter={}):
 109         pl_initscripts = {}
 110         filter.update({'enabled': True})
 111         for initscript in self.driver.shell.GetInitScripts(filter):
 112             pl_initscripts[initscript['initscript_id']] = initscript
 113         return pl_initscripts
 114
 115     def get_slivers(self, urns, options={}):
 116         names = set()
 117         slice_ids = set()
 118         node_ids = []
 119         for urn in urns:
 120             xrn = PlXrn(xrn=urn)
 121             if xrn.type == 'sliver':
 122                  # id: slice_id-node_id
 123                 sliver_id_parts = xrn.get_sliver_id_parts()
 124                 slice_ids.add(int(sliver_id_parts[0]))
 125                 node_ids.append(int(sliver_id_parts[1]))
 126             else:
 127                 names.add(xrn.pl_slicename())
 128
 129         filter = {}
 130         if names:
 131             filter['name'] = list(names)
 132         if slice_ids:
 133             filter['slice_id'] = list(slice_ids)
 134         slices = self.driver.shell.GetSlices(filter)
 135         if not slices:
 136             return []
 137         slice = slices[0]
 138         if node_ids:
 139             node_ids = [node_id for node_id in node_ids if node_id in slice['node_ids']]
 140             slice['node_ids'] = node_ids
 141         tags_dict = self.get_slice_tags(slice)
 142         nodes_dict = self.get_slice_nodes(slice, options)
 143         slivers = []
 144         for node in nodes_dict.values():
 145             node.update(slices[0])
 146             node['tags'] = tags_dict[node['node_id']]
 147             sliver_hrn = '%s.%s-%s' % (self.driver.hrn, slice['slice_id'], node['node_id'])
 148             node['sliver_id'] = Xrn(sliver_hrn, type='sliver').urn
 149             node['urn'] = node['sliver_id']
 150             slivers.append(node)
 151         return slivers
 152
 153     def node_to_rspec_node(self, node, sites, interfaces, node_tags, pl_initscripts=[], grain=None, options={}):
 154         rspec_node = Node()
 155         # xxx how to retrieve site['login_base']
 156         site=sites[node['site_id']]
 157         rspec_node['component_id'] = PlXrn(self.driver.hrn, hostname=node['hostname']).get_urn()
 158         rspec_node['component_name'] = node['hostname']
 159         rspec_node['component_manager_id'] = Xrn(self.driver.hrn, 'authority+cm').get_urn()
 160         rspec_node['authority_id'] = hrn_to_urn(PlXrn.site_hrn(self.driver.hrn, site['login_base']), 'authority+sa')
 161         # do not include boot state (<available> element) in the manifest rspec
 162         rspec_node['boot_state'] = node['boot_state']
 163         if node['boot_state'] == 'boot':
 164             rspec_node['available'] = 'true'
 165         else:
 166             rspec_node['available'] = 'false'
 167         rspec_node['exclusive'] = 'false'
 168         rspec_node['hardware_types'] = [HardwareType({'name': 'plab-pc'}),
 169                                         HardwareType({'name': 'pc'})]
 170         # only doing this because protogeni rspec needs
 171         # to advertise available initscripts
 172         rspec_node['pl_initscripts'] = pl_initscripts.values()
 173         # add site/interface info to nodes.
 174         # assumes that sites, interfaces and tags have already been prepared.
 175         if site['longitude'] and site['latitude']:
 176             location = Location({'longitude': site['longitude'], 'latitude': site['latitude'], 'country': 'unknown'})
 177             rspec_node['location'] = location
 178         # Granularity
 179         granularity = Granularity({'grain': grain})
 180         rspec_node['granularity'] = granularity
 181         rspec_node['interfaces'] = []
 182         if_count=0
 183         for if_id in node['interface_ids']:
 184             interface = Interface(interfaces[if_id])
 185             interface['ipv4'] = interface['ip']
 186             interface['component_id'] = PlXrn(auth=self.driver.hrn,
 187                                               interface='node%s:eth%s' % (node['node_id'], if_count)).get_urn()
 188             # interfaces in the manifest need a client id
 189             if slice:
 190                 interface['client_id'] = "%s:%s" % (node['node_id'], if_id)
 191             rspec_node['interfaces'].append(interface)
 192             if_count+=1
 193         tags = [PLTag(node_tags[tag_id]) for tag_id in node['node_tag_ids'] if tag_id in node_tags]
 194         rspec_node['tags'] = tags
 195         return rspec_node
 196
 197     def sliver_to_rspec_node(self, sliver, sites, interfaces, node_tags, \
 198                              pl_initscripts, sliver_allocations):
 199         # get the granularity in second for the reservation system
 200         grain = self.driver.shell.GetLeaseGranularity()
 201         rspec_node = self.node_to_rspec_node(sliver, sites, interfaces, node_tags, pl_initscripts, grain)
 202         # xxx how to retrieve site['login_base']
 203         rspec_node['expires'] = datetime_to_string(utcparse(sliver['expires']))
 204         # remove interfaces from manifest
 205         rspec_node['interfaces'] = []
 206         # add sliver info
 207         rspec_sliver = Sliver({'sliver_id': sliver['urn'],
 208                          'name': sliver['name'],
 209                          'type': 'plab-vserver',
 210                          'tags': []})
 211         rspec_node['sliver_id'] = rspec_sliver['sliver_id']
 212         rspec_node['client_id'] = sliver_allocations[sliver['urn']].client_id
 213         if sliver_allocations[sliver['urn']].component_id:
 214             rspec_node['component_id'] = sliver_allocations[sliver['urn']].component_id
 215         rspec_node['slivers'] = [rspec_sliver]
 216
 217         # slivers always provide the ssh service
 218         login = Login({'authentication': 'ssh-keys', 'hostname': sliver['hostname'], 'port':'22', 'username': sliver['name']})
 219         service = Services({'login': login})
 220         rspec_node['services'] = [service]
 221         return rspec_node
 222
 223     def get_slice_tags(self, slice):
 224         slice_tag_ids = []
 225         slice_tag_ids.extend(slice['slice_tag_ids'])
 226         tags = self.driver.shell.GetSliceTags({'slice_tag_id': slice_tag_ids})
 227         # sorted by node_id
 228         tags_dict = defaultdict(list)
 229         for tag in tags:
 230             tags_dict[tag['node_id']] = tag
 231         return tags_dict
 232
 233     def get_slice_nodes(self, slice, options={}):
 234         nodes_dict = {}
 235         filter = {'peer_id': None}
 236         tags_filter = {}
 237         if slice and slice.get('node_ids'):
 238             filter['node_id'] = slice['node_ids']
 239         else:
 240             # there are no nodes to look up
 241             return nodes_dict
 242         tags_filter=filter.copy()
 243         geni_available = options.get('geni_available')
 244         if geni_available == True:
 245             filter['boot_state'] = 'boot'
 246         nodes = self.driver.shell.GetNodes(filter)
 247         for node in nodes:
 248             nodes_dict[node['node_id']] = node
 249         return nodes_dict
 250
 251     def rspec_node_to_geni_sliver(self, rspec_node, sliver_allocations = {}):
 252         if rspec_node['sliver_id'] in sliver_allocations:
 253             # set sliver allocation and operational status
 254             sliver_allocation = sliver_allocations[rspec_node['sliver_id']]
 255             if sliver_allocation:
 256                 allocation_status = sliver_allocation.allocation_state
 257                 if allocation_status == 'geni_allocated':
 258                     op_status =  'geni_pending_allocation'
 259                 elif allocation_status == 'geni_provisioned':
 260                     if rspec_node['boot_state'] == 'boot':
 261                         op_status = 'geni_ready'
 262                     else:
 263                         op_status = 'geni_failed'
 264                 else:
 265                     op_status = 'geni_unknown'
 266             else:
 267                 allocation_status = 'geni_unallocated'
 268         # required fields
 269         geni_sliver = {'geni_sliver_urn': rspec_node['sliver_id'],
 270                        'geni_expires': rspec_node['expires'],
 271                        'geni_allocation_status' : allocation_status,
 272                        'geni_operational_status': op_status,
 273                        'geni_error': '',
 274                        }
 275         return geni_sliver
 276
 277     def get_leases(self, slice=None, options={}):
 278
 279         now = int(time.time())
 280         filter={}
 281         filter.update({'clip':now})
 282         if slice:
 283            filter.update({'name':slice['name']})
 284         return_fields = ['lease_id', 'hostname', 'site_id', 'name', 't_from', 't_until']
 285         leases = self.driver.shell.GetLeases(filter)
 286         grain = self.driver.shell.GetLeaseGranularity()
 287
 288         site_ids = []
 289         for lease in leases:
 290             site_ids.append(lease['site_id'])
 291
 292         # get sites
 293         sites_dict  = self.get_sites({'site_id': site_ids})
 294
 295         rspec_leases = []
 296         for lease in leases:
 297
 298             rspec_lease = Lease()
 299
 300             # xxx how to retrieve site['login_base']
 301             site_id=lease['site_id']
 302             site=sites_dict[site_id]
 303
 304             rspec_lease['lease_id'] = lease['lease_id']
 305             rspec_lease['component_id'] = PlXrn(self.driver.hrn, hostname=lease['hostname']).urn
 306             slice_hrn = slicename_to_hrn(self.driver.hrn, lease['name'])
 307             slice_urn = hrn_to_urn(slice_hrn, 'slice')
 308             rspec_lease['slice_id'] = slice_urn
 309             rspec_lease['start_time'] = lease['t_from']
 310             rspec_lease['duration'] = (lease['t_until'] - lease['t_from']) / grain
 311             rspec_leases.append(rspec_lease)
 312         return rspec_leases
 313
 314
 315     def list_resources(self, version = None, options={}):
 316
 317         version_manager = VersionManager()
 318         version = version_manager.get_version(version)
 319         rspec_version = version_manager._get_version(version.type, version.version, 'ad')
 320         rspec = RSpec(version=rspec_version, user_options=options)
 321
 322         if not options.get('list_leases') or options['list_leases'] != 'leases':
 323             # get nodes
 324             nodes  = self.get_nodes(options)
 325             site_ids = []
 326             interface_ids = []
 327             tag_ids = []
 328             nodes_dict = {}
 329             for node in nodes:
 330                 site_ids.append(node['site_id'])
 331                 interface_ids.extend(node['interface_ids'])
 332                 tag_ids.extend(node['node_tag_ids'])
 333                 nodes_dict[node['node_id']] = node
 334             sites = self.get_sites({'site_id': site_ids})
 335             interfaces = self.get_interfaces({'interface_id':interface_ids})
 336             node_tags = self.get_node_tags({'node_tag_id': tag_ids})
 337             pl_initscripts = self.get_pl_initscripts()
 338             # convert nodes to rspec nodes
 339             rspec_nodes = []
 340             for node in nodes:
 341                 rspec_node = self.node_to_rspec_node(node, sites, interfaces, node_tags, pl_initscripts)
 342                 rspec_nodes.append(rspec_node)
 343             rspec.version.add_nodes(rspec_nodes)
 344
 345             # add links
 346             links = self.get_links(sites, nodes_dict, interfaces)
 347             rspec.version.add_links(links)
 348         return rspec.toxml()
 349
 350     def describe(self, urns, version=None, options={}):
 351         version_manager = VersionManager()
 352         version = version_manager.get_version(version)
 353         rspec_version = version_manager._get_version(version.type, version.version, 'manifest')
 354         rspec = RSpec(version=rspec_version, user_options=options)
 355
 356         # get slivers
 357         geni_slivers = []
 358         slivers = self.get_slivers(urns, options)
 359         if slivers:
 360             rspec_expires = datetime_to_string(utcparse(slivers[0]['expires']))
 361         else:
 362             rspec_expires = datetime_to_string(utcparse(time.time()))
 363         rspec.xml.set('expires',  rspec_expires)
 364
 365         # lookup the sliver allocations
 366         sliver_ids = [sliver['sliver_id'] for sliver in slivers]
 367         constraint = SliverAllocation.sliver_id.in_(sliver_ids)
 368         sliver_allocations = dbsession.query(SliverAllocation).filter(constraint)
 369         sliver_allocation_dict = {}
 370         for sliver_allocation in sliver_allocations:
 371             sliver_allocation_dict[sliver_allocation.sliver_id] = sliver_allocation
 372
 373         if not options.get('list_leases') or options['list_leases'] != 'leases':
 374             # add slivers
 375             site_ids = []
 376             interface_ids = []
 377             tag_ids = []
 378             nodes_dict = {}
 379             for sliver in slivers:
 380                 site_ids.append(sliver['site_id'])
 381                 interface_ids.extend(sliver['interface_ids'])
 382                 tag_ids.extend(sliver['node_tag_ids'])
 383                 nodes_dict[sliver['node_id']] = sliver
 384             sites = self.get_sites({'site_id': site_ids})
 385             interfaces = self.get_interfaces({'interface_id':interface_ids})
 386             node_tags = self.get_node_tags({'node_tag_id': tag_ids})
 387             pl_initscripts = self.get_pl_initscripts()
 388             rspec_nodes = []
 389             for sliver in slivers:
 390                 if sliver['slice_ids_whitelist'] and sliver['slice_id'] not in sliver['slice_ids_whitelist']:
 391                     continue
 392                 rspec_node = self.sliver_to_rspec_node(sliver, sites, interfaces, node_tags,
 393                                                        pl_initscripts, sliver_allocation_dict)
 394                 # manifest node element shouldn't contain available attribute
 395                 rspec_node.pop('available')
 396                 rspec_nodes.append(rspec_node)
 397                 geni_sliver = self.rspec_node_to_geni_sliver(rspec_node, sliver_allocation_dict)
 398                 geni_slivers.append(geni_sliver)
 399             rspec.version.add_nodes(rspec_nodes)
 400
 401             # add sliver defaults
 402             #default_sliver = slivers.get(None, [])
 403             #if default_sliver:
 404             #    default_sliver_attribs = default_sliver.get('tags', [])
 405             #    for attrib in default_sliver_attribs:
 406             #        rspec.version.add_default_sliver_attribute(attrib['tagname'], attrib['value'])
 407
 408             # add links
 409             links = self.get_links(sites, nodes_dict, interfaces)
 410             rspec.version.add_links(links)
 411
 412         if not options.get('list_leases') or options['list_leases'] != 'resources':
 413             if slivers:
 414                 leases = self.get_leases(slivers[0])
 415                 rspec.version.add_leases(leases)
 416
 417
 418         return {'geni_urn': urns[0],
 419                 'geni_rspec': rspec.toxml(),
 420                 'geni_slivers': geni_slivers}