sfa/managers/aggregate/plc.py

   1 ### $Id: slices.py 15842 2009-11-22 09:56:13Z anil $
   2 ### $URL: https://svn.planet-lab.org/svn/sfa/trunk/sfa/plc/slices.py $
   3
   4 import datetime
   5 import time
   6 import traceback
   7 import sys
   8
   9 from types import StringTypes
  10 from sfa.util.misc import *
  11 from sfa.util.rspec import *
  12 from sfa.util.specdict import *
  13 from sfa.util.faults import *
  14 from sfa.util.storage import *
  15 from sfa.util.record import GeniRecord
  16 from sfa.util.policy import Policy
  17 from sfa.util.prefixTree import prefixTree
  18 from sfa.util.debug import log
  19 from sfa.server.registry import Registries
  20
  21 MAXINT =  2L**31-1
  22
  23 class Slices(SimpleStorage):
  24
  25     rspec_to_slice_tag = {'max_rate':'net_max_rate'}
  26
  27     def __init__(self, api, ttl = .5, caller_cred=None):
  28         self.api = api
  29         self.ttl = ttl
  30         self.threshold = None
  31         path = self.api.config.SFA_DATA_DIR
  32         filename = ".".join([self.api.interface, self.api.hrn, "slices"])
  33         filepath = path + os.sep + filename
  34         self.slices_file = filepath
  35         SimpleStorage.__init__(self, self.slices_file)
  36         self.policy = Policy(self.api)
  37         self.load()
  38         self.caller_cred=caller_cred
  39
  40     def get_slivers(self, hrn, node=None):
  41
  42         slice_name = hrn_to_pl_slicename(hrn)
  43         # XX Should we just call PLCAPI.GetSliceTicket(slice_name) instead
  44         # of doing all of this?
  45         #return self.api.GetSliceTicket(self.auth, slice_name)
  46
  47         # from PLCAPI.GetSlivers.get_slivers()
  48         slice_fields = ['slice_id', 'name', 'instantiation', 'expires', 'person_ids', 'slice_tag_ids']
  49         slices = self.api.plshell.GetSlices(self.api.plauth, slice_name, slice_fields)
  50         # Build up list of users and slice attributes
  51         person_ids = set()
  52         all_slice_tag_ids = set()
  53         for slice in slices:
  54             person_ids.update(slice['person_ids'])
  55             all_slice_tag_ids.update(slice['slice_tag_ids'])
  56         person_ids = list(person_ids)
  57         all_slice_tag_ids = list(all_slice_tag_ids)
  58         # Get user information
  59         all_persons_list = self.api.plshell.GetPersons(self.api.plauth, {'person_id':person_ids,'enabled':True}, ['person_id', 'enabled', 'key_ids'])
  60         all_persons = {}
  61         for person in all_persons_list:
  62             all_persons[person['person_id']] = person
  63
  64         # Build up list of keys
  65         key_ids = set()
  66         for person in all_persons.values():
  67             key_ids.update(person['key_ids'])
  68         key_ids = list(key_ids)
  69         # Get user account keys
  70         all_keys_list = self.api.plshell.GetKeys(self.api.plauth, key_ids, ['key_id', 'key', 'key_type'])
  71         all_keys = {}
  72         for key in all_keys_list:
  73             all_keys[key['key_id']] = key
  74         # Get slice attributes
  75         all_slice_tags_list = self.api.plshell.GetSliceTags(self.api.plauth, all_slice_tag_ids)
  76         all_slice_tags = {}
  77         for slice_tag in all_slice_tags_list:
  78             all_slice_tags[slice_tag['slice_tag_id']] = slice_tag
  79
  80         slivers = []
  81         for slice in slices:
  82             keys = []
  83             for person_id in slice['person_ids']:
  84                 if person_id in all_persons:
  85                     person = all_persons[person_id]
  86                     if not person['enabled']:
  87                         continue
  88                     for key_id in person['key_ids']:
  89                         if key_id in all_keys:
  90                             key = all_keys[key_id]
  91                             keys += [{'key_type': key['key_type'],
  92                                     'key': key['key']}]
  93             attributes = []
  94             # All (per-node and global) attributes for this slice
  95             slice_tags = []
  96             for slice_tag_id in slice['slice_tag_ids']:
  97                 if slice_tag_id in all_slice_tags:
  98                     slice_tags.append(all_slice_tags[slice_tag_id])
  99             # Per-node sliver attributes take precedence over global
 100             # slice attributes, so set them first.
 101             # Then comes nodegroup slice attributes
 102             # Followed by global slice attributes
 103             sliver_attributes = []
 104
 105             if node is not None:
 106                 for sliver_attribute in filter(lambda a: a['node_id'] == node['node_id'], slice_tags):
 107                     sliver_attributes.append(sliver_attribute['tagname'])
 108                     attributes.append({'tagname': sliver_attribute['tagname'],
 109                                     'value': sliver_attribute['value']})
 110
 111             # set nodegroup slice attributes
 112             for slice_tag in filter(lambda a: a['nodegroup_id'] in node['nodegroup_ids'], slice_tags):
 113                 # Do not set any nodegroup slice attributes for
 114                 # which there is at least one sliver attribute
 115                 # already set.
 116                 if slice_tag not in slice_tags:
 117                     attributes.append({'tagname': slice_tag['tagname'],
 118                         'value': slice_tag['value']})
 119
 120             for slice_tag in filter(lambda a: a['node_id'] is None, slice_tags):
 121                 # Do not set any global slice attributes for
 122                 # which there is at least one sliver attribute
 123                 # already set.
 124                 if slice_tag['tagname'] not in sliver_attributes:
 125                     attributes.append({'tagname': slice_tag['tagname'],
 126                                    'value': slice_tag['value']})
 127
 128             # XXX Sanity check; though technically this should be a system invariant
 129             # checked with an assertion
 130             if slice['expires'] > MAXINT:  slice['expires']= MAXINT
 131
 132             slivers.append({
 133                 'hrn': hrn,
 134                 'name': slice['name'],
 135                 'slice_id': slice['slice_id'],
 136                 'instantiation': slice['instantiation'],
 137                 'expires': slice['expires'],
 138                 'keys': keys,
 139                 'attributes': attributes
 140             })
 141
 142         return slivers
 143
 144     def get_peer(self, hrn):
 145         # Becaues of myplc federation,  we first need to determine if this
 146         # slice belongs to out local plc or a myplc peer. We will assume it
 147         # is a local site, unless we find out otherwise
 148         peer = None
 149
 150         # get this slice's authority (site)
 151         slice_authority = get_authority(hrn)
 152
 153         # get this site's authority (sfa root authority or sub authority)
 154         site_authority = get_authority(slice_authority).lower()
 155
 156         # check if we are already peered with this site_authority, if so
 157         peers = self.api.plshell.GetPeers(self.api.plauth, {}, ['peer_id', 'peername', 'shortname', 'hrn_root'])
 158         for peer_record in peers:
 159             names = [name.lower() for name in peer_record.values() if isinstance(name, StringTypes)]
 160             if site_authority in names:
 161                 peer = peer_record['shortname']
 162
 163         return peer
 164
 165     def get_sfa_peer(self, hrn):
 166         # return the authority for this hrn or None if we are the authority
 167         sfa_peer = None
 168         slice_authority = get_authority(hrn)
 169         site_authority = get_authority(slice_authority)
 170
 171         if site_authority != self.api.hrn:
 172             sfa_peer = site_authority
 173
 174         return sfa_peer
 175
 176     def refresh(self):
 177         """
 178         Update the cached list of slices
 179         """
 180         # Reload components list
 181         now = datetime.datetime.now()
 182         if not self.has_key('threshold') or not self.has_key('timestamp') or \
 183            now > datetime.datetime.fromtimestamp(time.mktime(time.strptime(self['threshold'], self.api.time_format))):
 184            self.refresh_slices_aggregate()
 185
 186     def refresh_slices_aggregate(self):
 187         slices = self.api.plshell.GetSlices(self.api.plauth, {'peer_id': None}, ['name'])
 188         slice_hrns = [slicename_to_hrn(self.api.hrn, slice['name']) for slice in slices]
 189
 190          # update timestamp and threshold
 191         timestamp = datetime.datetime.now()
 192         hr_timestamp = timestamp.strftime(self.api.time_format)
 193         delta = datetime.timedelta(hours=self.ttl)
 194         threshold = timestamp + delta
 195         hr_threshold = threshold.strftime(self.api.time_format)
 196
 197         slice_details = {'hrn': slice_hrns,
 198                          'timestamp': hr_timestamp,
 199                          'threshold': hr_threshold
 200                         }
 201         self.update(slice_details)
 202         self.write()
 203
 204
 205     def delete_slice(self, hrn):
 206         self.delete_slice_aggregate(hrn)
 207
 208     def delete_slice_aggregate(self, hrn):
 209
 210         slicename = hrn_to_pl_slicename(hrn)
 211         slices = self.api.plshell.GetSlices(self.api.plauth, {'name': slicename})
 212         if not slices:
 213             return 1
 214         slice = slices[0]
 215
 216         # determine if this is a peer slice
 217         peer = self.get_peer(hrn)
 218         if peer:
 219             self.api.plshell.UnBindObjectFromPeer(self.api.plauth, 'slice', slice['slice_id'], peer)
 220         self.api.plshell.DeleteSliceFromNodes(self.api.plauth, slicename, slice['node_ids'])
 221         if peer:
 222             self.api.plshell.BindObjectToPeer(self.api.plauth, 'slice', slice['slice_id'], peer, slice['peer_slice_id'])
 223         return 1
 224
 225     def verify_site(self, registry, credential, slice_hrn, peer, sfa_peer):
 226         authority = get_authority(slice_hrn)
 227         try:
 228             site_records = registry.resolve(credential, authority)
 229         except:
 230             arg_list = [credential, authority]
 231             request_hash = self.api.key.compute_hash(arg_list)
 232             site_records = registry.resolve(credential, authority, request_hash)
 233
 234         site = {}
 235         for site_record in site_records:
 236             if site_record['type'] == 'authority':
 237                 site = site_record
 238         if not site:
 239             raise RecordNotFound(authority)
 240         remote_site_id = site.pop('site_id')
 241
 242         login_base = get_leaf(authority)
 243         sites = self.api.plshell.GetSites(self.api.plauth, login_base)
 244         if not sites:
 245             site_id = self.api.plshell.AddSite(self.api.plauth, site)
 246             if peer:
 247                 self.api.plshell.BindObjectToPeer(self.api.plauth, 'site', site_id, peer, remote_site_id)
 248             # mark this site as an sfa peer record
 249             if sfa_peer:
 250                 peer_dict = {'type': 'authority', 'hrn': authority, 'peer_authority': sfa_peer, 'pointer': site_id}
 251                 try:
 252                     registry.register_peer_object(credential, peer_dict)
 253                 except:
 254                     arg_list = [credential]
 255                     request_hash = self.api.key.compute_hash(arg_list)
 256                     registry.register_peer_object(credential, peer_dict, request_hash)
 257         else:
 258             site_id = sites[0]['site_id']
 259             remote_site_id = sites[0]['peer_site_id']
 260
 261
 262         return (site_id, remote_site_id)
 263
 264     def verify_slice(self, registry, credential, slice_hrn, site_id, remote_site_id, peer, sfa_peer):
 265         slice = {}
 266         slice_record = None
 267         authority = get_authority(slice_hrn)
 268         try:
 269             slice_records = registry.resolve(credential, slice_hrn)
 270         except:
 271             arg_list = [credential, slice_hrn]
 272             request_hash = self.api.key.compute_hash(arg_list)
 273             slice_records = registry.resolve(credential, slice_hrn, request_hash)
 274
 275         for record in slice_records:
 276             if record['type'] in ['slice']:
 277                 slice_record = record
 278         if not slice_record:
 279             raise RecordNotFound(hrn)
 280         slicename = hrn_to_pl_slicename(slice_hrn)
 281         parts = slicename.split("_")
 282         login_base = parts[0]
 283         slices = self.api.plshell.GetSlices(self.api.plauth, [slicename], ['slice_id', 'node_ids', 'site_id'])
 284         if not slices:
 285             slice_fields = {}
 286             slice_keys = ['name', 'url', 'description']
 287             for key in slice_keys:
 288                 if key in slice_record and slice_record[key]:
 289                     slice_fields[key] = slice_record[key]
 290
 291             # add the slice
 292             slice_id = self.api.plshell.AddSlice(self.api.plauth, slice_fields)
 293             slice = slice_fields
 294             slice['slice_id'] = slice_id
 295
 296             # mark this slice as an sfa peer record
 297             if sfa_peer:
 298                 peer_dict = {'type': 'slice', 'hrn': slice_hrn, 'peer_authority': sfa_peer, 'pointer': slice_id}
 299                 try:
 300                     registry.register_peer_object(credential, peer_dict)
 301                 except:
 302                     arg_list = [credential]
 303                     request_hash = self.api.key.compute_hash(arg_list)
 304                     registry.register_peer_object(credential, peer_dict, request_hash)
 305
 306             #this belongs to a peer
 307             if peer:
 308                 self.api.plshell.BindObjectToPeer(self.api.plauth, 'slice', slice_id, peer, slice_record['pointer'])
 309             slice['node_ids'] = []
 310         else:
 311             slice = slices[0]
 312             slice_id = slice['slice_id']
 313             site_id = slice['site_id']
 314
 315         slice['peer_slice_id'] = slice_record['pointer']
 316         self.verify_persons(registry, credential, slice_record, site_id, remote_site_id, peer, sfa_peer)
 317
 318         return slice
 319
 320     def verify_persons(self, registry, credential, slice_record, site_id, remote_site_id, peer, sfa_peer):
 321         # get the list of valid slice users from the registry and make
 322         # sure they are added to the slice
 323         slicename = hrn_to_pl_slicename(slice_record['hrn'])
 324         researchers = slice_record.get('researcher', [])
 325         for researcher in researchers:
 326             person_record = {}
 327             try:
 328                 person_records = registry.resolve(credential, researcher)
 329             except:
 330                 arg_list = [credential, researcher]
 331                 request_hash = self.api.key.compute_hash(arg_list)
 332                 person_records = registry.resolve(credential, researcher, request_hash)
 333             for record in person_records:
 334                 if record['type'] in ['user']:
 335                     person_record = record
 336             if not person_record:
 337                 pass
 338             person_dict = person_record
 339             local_person=False
 340             if peer:
 341                 peer_id = self.api.plshell.GetPeers(self.api.plauth, {'shortname': peer}, ['peer_id'])[0]['peer_id']
 342                 persons = self.api.plshell.GetPersons(self.api.plauth, {'email': [person_dict['email']], 'peer_id': peer_id}, ['person_id', 'key_ids'])
 343                 if not persons:
 344                     persons = self.api.plshell.GetPersons(self.api.plauth, [person_dict['email']], ['person_id', 'key_ids'])
 345                     if persons:
 346                        local_person=True
 347
 348             else:
 349                 persons = self.api.plshell.GetPersons(self.api.plauth, [person_dict['email']], ['person_id', 'key_ids'])
 350
 351             if not persons:
 352                 person_id=self.api.plshell.AddPerson(self.api.plauth, person_dict)
 353                 self.api.plshell.UpdatePerson(self.api.plauth, person_id, {'enabled' : True})
 354
 355                 # mark this person as an sfa peer record
 356                 if sfa_peer:
 357                     peer_dict = {'type': 'user', 'hrn': researcher, 'peer_authority': sfa_peer, 'pointer': person_id}
 358                     try:
 359                         registry.register_peer_object(credential, peer_dict)
 360                     except:
 361                         arg_list = [credential]
 362                         request_hash = self.api.key.compute_hash(arg_list)
 363                         registry.register_peer_object(credential, peer_dict, request_hash)
 364
 365                 if peer:
 366                     self.api.plshell.BindObjectToPeer(self.api.plauth, 'person', person_id, peer, person_dict['pointer'])
 367                 key_ids = []
 368             else:
 369                 person_id = persons[0]['person_id']
 370                 key_ids = persons[0]['key_ids']
 371
 372
 373             # if this is a peer person, we must unbind them from the peer or PLCAPI will throw
 374             # an error
 375             if peer:
 376                 self.api.plshell.UnBindObjectFromPeer(self.api.plauth, 'person', person_id, peer)
 377                 self.api.plshell.UnBindObjectFromPeer(self.api.plauth, 'site', site_id,  peer)
 378
 379             self.api.plshell.AddPersonToSlice(self.api.plauth, person_dict['email'], slicename)
 380             self.api.plshell.AddPersonToSite(self.api.plauth, person_dict['email'], site_id)
 381             if peer and not local_person:
 382                 self.api.plshell.BindObjectToPeer(self.api.plauth, 'person', person_id, peer, person_dict['pointer'])
 383             if peer:
 384                 self.api.plshell.BindObjectToPeer(self.api.plauth, 'site', site_id, peer, remote_site_id)
 385
 386             self.verify_keys(registry, credential, person_dict, key_ids, person_id, peer, local_person)
 387
 388     def verify_keys(self, registry, credential, person_dict, key_ids, person_id,  peer, local_person):
 389         keylist = self.api.plshell.GetKeys(self.api.plauth, key_ids, ['key'])
 390         keys = [key['key'] for key in keylist]
 391
 392         #add keys that arent already there
 393         key_ids = person_dict['key_ids']
 394         for personkey in person_dict['keys']:
 395             if personkey not in keys:
 396                 key = {'key_type': 'ssh', 'key': personkey}
 397                 if peer:
 398                     self.api.plshell.UnBindObjectFromPeer(self.api.plauth, 'person', person_id, peer)
 399                 key_id = self.api.plshell.AddPersonKey(self.api.plauth, person_dict['email'], key)
 400                 if peer and not local_person:
 401                     self.api.plshell.BindObjectToPeer(self.api.plauth, 'person', person_id, peer, person_dict['pointer'])
 402                 if peer:
 403                     try: self.api.plshell.BindObjectToPeer(self.api.plauth, 'key', key_id, peer, key_ids.pop(0))
 404
 405                     except: pass
 406
 407     def create_slice(self, hrn, rspec):
 408
 409         # check our slice policy before we procede
 410         whitelist = self.policy['slice_whitelist']
 411         blacklist = self.policy['slice_blacklist']
 412
 413         if whitelist and hrn not in whitelist or \
 414            blacklist and hrn in blacklist:
 415             policy_file = self.policy.policy_file
 416             print >> log, "Slice %(hrn)s not allowed by policy %(policy_file)s" % locals()
 417             return 1
 418
 419         self.create_slice_aggregate(hrn, rspec)
 420
 421     def create_slice_aggregate(self, hrn, rspec):
 422
 423         # Determine if this is a peer slice
 424         peer = self.get_peer(hrn)
 425         sfa_peer = self.get_sfa_peer(hrn)
 426
 427         spec = RSpec(rspec)
 428         # Get the slice record from sfa
 429         slicename = hrn_to_pl_slicename(hrn)
 430         slice = {}
 431         slice_record = None
 432         registries = Registries(self.api)
 433         registry = registries[self.api.hrn]
 434         credential = self.api.getCredential()
 435
 436         site_id, remote_site_id = self.verify_site(registry, credential, hrn, peer, sfa_peer)
 437         slice = self.verify_slice(registry, credential, hrn, site_id, remote_site_id, peer, sfa_peer)
 438
 439         # find out where this slice is currently running
 440         nodelist = self.api.plshell.GetNodes(self.api.plauth, slice['node_ids'], ['hostname'])
 441         hostnames = [node['hostname'] for node in nodelist]
 442
 443         # get netspec details
 444         nodespecs = spec.getDictsByTagName('NodeSpec')
 445
 446         # dict in which to store slice attributes to set for the nodes
 447         nodes = {}
 448         for nodespec in nodespecs:
 449             if isinstance(nodespec['name'], list):
 450                 for nodename in nodespec['name']:
 451                     nodes[nodename] = {}
 452                     for k in nodespec.keys():
 453                         rspec_attribute_value = nodespec[k]
 454                         if (self.rspec_to_slice_tag.has_key(k)):
 455                             slice_tag_name = self.rspec_to_slice_tag[k]
 456                             nodes[nodename][slice_tag_name] = rspec_attribute_value
 457             elif isinstance(nodespec['name'], StringTypes):
 458                 nodename = nodespec['name']
 459                 nodes[nodename] = {}
 460                 for k in nodespec.keys():
 461                     rspec_attribute_value = nodespec[k]
 462                     if (self.rspec_to_slice_tag.has_key(k)):
 463                         slice_tag_name = self.rspec_to_slice_tag[k]
 464                         nodes[nodename][slice_tag_name] = rspec_attribute_value
 465
 466                 for k in nodespec.keys():
 467                     rspec_attribute_value = nodespec[k]
 468                     if (self.rspec_to_slice_tag.has_key(k)):
 469                         slice_tag_name = self.rspec_to_slice_tag[k]
 470                         nodes[nodename][slice_tag_name] = rspec_attribute_value
 471
 472         node_names = nodes.keys()
 473         # remove nodes not in rspec
 474         deleted_nodes = list(set(hostnames).difference(node_names))
 475         # add nodes from rspec
 476         added_nodes = list(set(node_names).difference(hostnames))
 477
 478         if peer:
 479             self.api.plshell.UnBindObjectFromPeer(self.api.plauth, 'slice', slice['slice_id'], peer)
 480
 481         self.api.plshell.AddSliceToNodes(self.api.plauth, slicename, added_nodes)
 482
 483         # Add recognized slice tags
 484         for node_name in node_names:
 485             node = nodes[node_name]
 486             for slice_tag in node.keys():
 487                 value = node[slice_tag]
 488                 if (isinstance(value, list)):
 489                     value = value[0]
 490
 491                 self.api.plshell.AddSliceTag(self.api.plauth, slicename, slice_tag, value, node_name)
 492
 493         self.api.plshell.DeleteSliceFromNodes(self.api.plauth, slicename, deleted_nodes)
 494         if peer:
 495             self.api.plshell.BindObjectToPeer(self.api.plauth, 'slice', slice['slice_id'], peer, slice['peer_slice_id'])
 496
 497         return 1
 498
 499     def start_slice(self, hrn):
 500         self.start_slice_aggregate(hrn)
 501
 502     def start_slice_aggregate(self, hrn):
 503         slicename = hrn_to_pl_slicename(hrn)
 504         slices = self.api.plshell.GetSlices(self.api.plauth, {'name': slicename}, ['slice_id'])
 505         if not slices:
 506             raise RecordNotFound(hrn)
 507         slice_id = slices[0]
 508         attributes = self.api.plshell.GetSliceAttributes(self.api.plauth, {'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 509         attribute_id = attreibutes[0]['slice_attribute_id']
 510         self.api.plshell.UpdateSliceAttribute(self.api.plauth, attribute_id, "1" )
 511         return 1
 512
 513
 514     def stop_slice(self, hrn):
 515         self.stop_slice_aggregate(hrn)
 516
 517     def stop_slice_aggregate(self, hrn):
 518         slicename = hrn_to_pl_slicename(hrn)
 519         slices = self.api.plshell.GetSlices(self.api.plauth, {'name': slicename}, ['slice_id'])
 520         if not slices:
 521             raise RecordNotFound(hrn)
 522         slice_id = slices[0]['slice_id']
 523         attributes = self.api.plshell.GetSliceAttributes(self.api.plauth, {'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 524         attribute_id = attributes[0]['slice_attribute_id']
 525         self.api.plshell.UpdateSliceAttribute(self.api.plauth, attribute_id, "0")
 526         return 1
 527