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