sfa/dummy/dummydriver.py

   1 import time
   2 import datetime
   3 #
   4 from sfa.util.faults import MissingSfaInfo, UnknownSfaType, \
   5     RecordNotFound, SfaNotImplemented, SliverDoesNotExist, SearchFailed, \
   6     UnsupportedOperation, Forbidden
   7 from sfa.util.sfalogging import logger
   8 from sfa.util.defaultdict import defaultdict
   9 from sfa.util.sfatime import utcparse, datetime_to_string, datetime_to_epoch
  10 from sfa.util.xrn import Xrn, hrn_to_urn, get_leaf
  11 from sfa.util.cache import Cache
  12
  13 # one would think the driver should not need to mess with the SFA db, but..
  14 from sfa.storage.model import RegRecord, SliverAllocation
  15 from sfa.trust.credential import Credential
  16
  17 # used to be used in get_ticket
  18 #from sfa.trust.sfaticket import SfaTicket
  19 from sfa.rspecs.version_manager import VersionManager
  20 from sfa.rspecs.rspec import RSpec
  21
  22 # the driver interface, mostly provides default behaviours
  23 from sfa.managers.driver import Driver
  24
  25 from sfa.dummy.dummyshell import DummyShell
  26 from sfa.dummy.dummyaggregate import DummyAggregate
  27 from sfa.dummy.dummyslices import DummySlices
  28 from sfa.dummy.dummyxrn import DummyXrn, slicename_to_hrn, hostname_to_hrn, hrn_to_dummy_slicename, xrn_to_hostname
  29
  30
  31 def list_to_dict(recs, key):
  32     """
  33     convert a list of dictionaries into a dictionary keyed on the
  34     specified dictionary key
  35     """
  36     return dict ( [ (rec[key],rec) for rec in recs ] )
  37
  38 #
  39 # DummyShell is just an xmlrpc serverproxy where methods can be sent as-is;
  40 #
  41 class DummyDriver (Driver):
  42
  43     # the cache instance is a class member so it survives across incoming requests
  44     cache = None
  45
  46     def __init__ (self, api):
  47         Driver.__init__ (self, api)
  48         config = api.config
  49         self.hrn = config.SFA_INTERFACE_HRN
  50         self.root_auth = config.SFA_REGISTRY_ROOT_AUTH
  51         self.shell = DummyShell (config)
  52         self.testbedInfo = self.shell.GetTestbedInfo()
  53
  54     def check_sliver_credentials(self, creds, urns):
  55         # build list of cred object hrns
  56         slice_cred_names = []
  57         for cred in creds:
  58             slice_cred_hrn = Credential(cred=cred).get_gid_object().get_hrn()
  59             slice_cred_names.append(DummyXrn(xrn=slice_cred_hrn).dummy_slicename())
  60
  61         # look up slice name of slivers listed in urns arg
  62         slice_ids = []
  63         for urn in urns:
  64             sliver_id_parts = Xrn(xrn=urn).get_sliver_id_parts()
  65             try:
  66                 slice_ids.append(int(sliver_id_parts[0]))
  67             except ValueError:
  68                 pass
  69
  70         if not slice_ids:
  71              raise Forbidden("sliver urn not provided")
  72
  73         slices = self.shell.GetSlices({'slice_ids': slice_ids})
  74         sliver_names = [slice['slice_name'] for slice in slices]
  75
  76         # make sure we have a credential for every specified sliver ierd
  77         for sliver_name in sliver_names:
  78             if sliver_name not in slice_cred_names:
  79                 msg = "Valid credential not found for target: %s" % sliver_name
  80                 raise Forbidden(msg)
  81
  82     ########################################
  83     ########## registry oriented
  84     ########################################
  85
  86     def augment_records_with_testbed_info (self, sfa_records):
  87         return self.fill_record_info (sfa_records)
  88
  89     ##########
  90     def register (self, sfa_record, hrn, pub_key):
  91         type = sfa_record['type']
  92         dummy_record = self.sfa_fields_to_dummy_fields(type, hrn, sfa_record)
  93
  94         if type == 'authority':
  95             pointer = -1
  96
  97         elif type == 'slice':
  98             slices = self.shell.GetSlices({'slice_name': dummy_record['slice_name']})
  99             if not slices:
 100                  pointer = self.shell.AddSlice(dummy_record)
 101             else:
 102                  pointer = slices[0]['slice_id']
 103
 104         elif type == 'user':
 105             users = self.shell.GetUsers({'email':sfa_record['email']})
 106             if not users:
 107                 pointer = self.shell.AddUser(dummy_record)
 108             else:
 109                 pointer = users[0]['user_id']
 110
 111             # Add the user's key
 112             if pub_key:
 113                 self.shell.AddUserKey({'user_id' : pointer, 'key' : pub_key})
 114
 115         elif type == 'node':
 116             nodes = self.shell.GetNodes(dummy_record['hostname'])
 117             if not nodes:
 118                 pointer = self.shell.AddNode(dummy_record)
 119             else:
 120                 pointer = users[0]['node_id']
 121
 122         return pointer
 123
 124     ##########
 125     def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
 126         pointer = old_sfa_record['pointer']
 127         type = old_sfa_record['type']
 128         dummy_record=self.sfa_fields_to_dummy_fields(type, hrn, new_sfa_record)
 129
 130         # new_key implemented for users only
 131         if new_key and type not in [ 'user' ]:
 132             raise UnknownSfaType(type)
 133
 134
 135         if type == "slice":
 136             self.shell.UpdateSlice({'slice_id': pointer, 'fields': dummy_record})
 137
 138         elif type == "user":
 139             self.shell.UpdateUser({'user_id': pointer, 'fields': dummy_record})
 140
 141             if new_key:
 142                 self.shell.AddUserKey({'user_id' : pointer, 'key' : new_key})
 143
 144         elif type == "node":
 145             self.shell.UpdateNode({'node_id': pointer, 'fields': dummy_record})
 146
 147
 148         return True
 149
 150
 151     ##########
 152     def remove (self, sfa_record):
 153         type=sfa_record['type']
 154         pointer=sfa_record['pointer']
 155         if type == 'user':
 156             self.shell.DeleteUser({'user_id': pointer})
 157         elif type == 'slice':
 158             self.shell.DeleteSlice({'slice_id': pointer})
 159         elif type == 'node':
 160             self.shell.DeleteNode({'node_id': pointer})
 161
 162         return True
 163
 164
 165
 166
 167
 168     ##
 169     # Convert SFA fields to Dummy testbed fields for use when registering or updating
 170     # registry record in the dummy testbed
 171     #
 172
 173     def sfa_fields_to_dummy_fields(self, type, hrn, sfa_record):
 174
 175         dummy_record = {}
 176
 177         if type == "slice":
 178             dummy_record["slice_name"] = hrn_to_dummy_slicename(hrn)
 179
 180         elif type == "node":
 181             if "hostname" not in sfa_record:
 182                 raise MissingSfaInfo("hostname")
 183             dummy_record["hostname"] = sfa_record["hostname"]
 184             if "type" in sfa_record:
 185                dummy_record["type"] = sfa_record["type"]
 186             else:
 187                dummy_record["type"] = "dummy_type"
 188
 189         elif type == "authority":
 190             dummy_record["name"] = hrn
 191
 192         elif type == "user":
 193             dummy_record["user_name"] = sfa_record["email"].split('@')[0]
 194             dummy_record["email"] = sfa_record["email"]
 195
 196         return dummy_record
 197
 198     ####################
 199     def fill_record_info(self, records):
 200         """
 201         Given a (list of) SFA record, fill in the DUMMY TESTBED specific
 202         and SFA specific fields in the record.
 203         """
 204         if not isinstance(records, list):
 205             records = [records]
 206
 207         self.fill_record_dummy_info(records)
 208         self.fill_record_hrns(records)
 209         self.fill_record_sfa_info(records)
 210         return records
 211
 212     def fill_record_dummy_info(self, records):
 213         """
 214         Fill in the DUMMY specific fields of a SFA record. This
 215         involves calling the appropriate DUMMY method to retrieve the
 216         database record for the object.
 217
 218         @param record: record to fill in field (in/out param)
 219         """
 220         # get ids by type
 221         node_ids, slice_ids, user_ids = [], [], []
 222         type_map = {'node': node_ids, 'slice': slice_ids, 'user': user_ids}
 223
 224         for record in records:
 225             for type in type_map:
 226                 if type == record['type']:
 227                     type_map[type].append(record['pointer'])
 228
 229         # get dummy records
 230         nodes, slices, users = {}, {}, {}
 231         if node_ids:
 232             node_list = self.shell.GetNodes({'node_ids':node_ids})
 233             nodes = list_to_dict(node_list, 'node_id')
 234         if slice_ids:
 235             slice_list = self.shell.GetSlices({'slice_ids':slice_ids})
 236             slices = list_to_dict(slice_list, 'slice_id')
 237         if user_ids:
 238             user_list = self.shell.GetUsers({'user_ids': user_ids})
 239             users = list_to_dict(user_list, 'user_id')
 240
 241         dummy_records = {'node': nodes, 'slice': slices, 'user': users}
 242
 243
 244         # fill record info
 245         for record in records:
 246             # records with pointer==-1 do not have dummy info.
 247             if record['pointer'] == -1:
 248                 continue
 249
 250             for type in dummy_records:
 251                 if record['type'] == type:
 252                     if record['pointer'] in dummy_records[type]:
 253                         record.update(dummy_records[type][record['pointer']])
 254                         break
 255             # fill in key info
 256             if record['type'] == 'user':
 257                 record['key_ids'] = []
 258                 record['keys'] = []
 259                 for key in dummy_records['user'][record['pointer']]['keys']:
 260                      record['key_ids'].append(-1)
 261                      record['keys'].append(key)
 262
 263         return records
 264
 265     def fill_record_hrns(self, records):
 266         """
 267         convert dummy ids to hrns
 268         """
 269
 270         # get ids
 271         slice_ids, user_ids, node_ids = [], [], []
 272         for record in records:
 273             if 'user_ids' in record:
 274                 user_ids.extend(record['user_ids'])
 275             if 'slice_ids' in record:
 276                 slice_ids.extend(record['slice_ids'])
 277             if 'node_ids' in record:
 278                 node_ids.extend(record['node_ids'])
 279
 280         # get dummy records
 281         slices, users, nodes = {}, {}, {}
 282         if user_ids:
 283             user_list = self.shell.GetUsers({'user_ids': user_ids})
 284             users = list_to_dict(user_list, 'user_id')
 285         if slice_ids:
 286             slice_list = self.shell.GetSlices({'slice_ids': slice_ids})
 287             slices = list_to_dict(slice_list, 'slice_id')
 288         if node_ids:
 289             node_list = self.shell.GetNodes({'node_ids': node_ids})
 290             nodes = list_to_dict(node_list, 'node_id')
 291
 292         # convert ids to hrns
 293         for record in records:
 294             # get all relevant data
 295             type = record['type']
 296             pointer = record['pointer']
 297             testbed_name = self.testbed_name()
 298             auth_hrn = self.hrn
 299             if pointer == -1:
 300                 continue
 301
 302             if 'user_ids' in record:
 303                 emails = [users[user_id]['email'] for user_id in record['user_ids'] \
 304                           if user_id in  users]
 305                 usernames = [email.split('@')[0] for email in emails]
 306                 user_hrns = [".".join([auth_hrn, testbed_name, username]) for username in usernames]
 307                 record['users'] = user_hrns
 308             if 'slice_ids' in record:
 309                 slicenames = [slices[slice_id]['slice_name'] for slice_id in record['slice_ids'] \
 310                               if slice_id in slices]
 311                 slice_hrns = [slicename_to_hrn(auth_hrn, slicename) for slicename in slicenames]
 312                 record['slices'] = slice_hrns
 313             if 'node_ids' in record:
 314                 hostnames = [nodes[node_id]['hostname'] for node_id in record['node_ids'] \
 315                              if node_id in nodes]
 316                 node_hrns = [hostname_to_hrn(auth_hrn, login_base, hostname) for hostname in hostnames]
 317                 record['nodes'] = node_hrns
 318
 319
 320         return records
 321
 322     def fill_record_sfa_info(self, records):
 323
 324         def startswith(prefix, values):
 325             return [value for value in values if value.startswith(prefix)]
 326
 327         # get user ids
 328         user_ids = []
 329         for record in records:
 330             user_ids.extend(record.get("user_ids", []))
 331
 332         # get sfa records for all records associated with these records.
 333         # we'll replace pl ids (person_ids) with hrns from the sfa records
 334         # we obtain
 335
 336         # get the registry records
 337         user_list, users = [], {}
 338         user_list = self.api.dbsession().query (RegRecord).filter(RegRecord.pointer.in_(user_ids))
 339         # create a hrns keyed on the sfa record's pointer.
 340         # Its possible for multiple records to have the same pointer so
 341         # the dict's value will be a list of hrns.
 342         users = defaultdict(list)
 343         for user in user_list:
 344             users[user.pointer].append(user)
 345
 346         # get the dummy records
 347         dummy_user_list, dummy_users = [], {}
 348         dummy_user_list = self.shell.GetUsers({'user_ids': user_ids})
 349         dummy_users = list_to_dict(dummy_user_list, 'user_id')
 350
 351         # fill sfa info
 352         for record in records:
 353             # skip records with no pl info (top level authorities)
 354             #if record['pointer'] == -1:
 355             #    continue
 356             sfa_info = {}
 357             type = record['type']
 358             logger.info("fill_record_sfa_info - incoming record typed %s"%type)
 359             if (type == "slice"):
 360                 # all slice users are researchers
 361                 record['geni_urn'] = hrn_to_urn(record['hrn'], 'slice')
 362                 record['PI'] = []
 363                 record['researcher'] = []
 364                 for user_id in record.get('user_ids', []):
 365                     hrns = [user.hrn for user in users[user_id]]
 366                     record['researcher'].extend(hrns)
 367
 368             elif (type.startswith("authority")):
 369                 record['url'] = None
 370                 logger.info("fill_record_sfa_info - authority xherex")
 371
 372             elif (type == "node"):
 373                 sfa_info['dns'] = record.get("hostname", "")
 374                 # xxx TODO: URI, LatLong, IP, DNS
 375
 376             elif (type == "user"):
 377                 logger.info('setting user.email')
 378                 sfa_info['email'] = record.get("email", "")
 379                 sfa_info['geni_urn'] = hrn_to_urn(record['hrn'], 'user')
 380                 sfa_info['geni_certificate'] = record['gid']
 381                 # xxx TODO: PostalAddress, Phone
 382             record.update(sfa_info)
 383
 384
 385     ####################
 386     def update_relation (self, subject_type, target_type, relation_name, subject_id, target_ids):
 387         # hard-wire the code for slice/user for now, could be smarter if needed
 388         if subject_type =='slice' and target_type == 'user' and relation_name == 'researcher':
 389             subject=self.shell.GetSlices ({'slice_id': subject_id})[0]
 390             if 'user_ids' not in subject.keys():
 391                  subject['user_ids'] = []
 392             current_target_ids = subject['user_ids']
 393             add_target_ids = list ( set (target_ids).difference(current_target_ids))
 394             del_target_ids = list ( set (current_target_ids).difference(target_ids))
 395             logger.debug ("subject_id = %s (type=%s)"%(subject_id,type(subject_id)))
 396             for target_id in add_target_ids:
 397                 self.shell.AddUserToSlice ({'user_id': target_id, 'slice_id': subject_id})
 398                 logger.debug ("add_target_id = %s (type=%s)"%(target_id,type(target_id)))
 399             for target_id in del_target_ids:
 400                 logger.debug ("del_target_id = %s (type=%s)"%(target_id,type(target_id)))
 401                 self.shell.DeleteUserFromSlice ({'user_id': target_id, 'slice_id': subject_id})
 402         else:
 403             logger.info('unexpected relation %s to maintain, %s -> %s'%(relation_name,subject_type,target_type))
 404
 405
 406     ########################################
 407     ########## aggregate oriented
 408     ########################################
 409
 410     def testbed_name (self): return "dummy"
 411
 412     def aggregate_version (self):
 413         return {}
 414
 415     def list_resources (self, version=None, options={}):
 416         aggregate = DummyAggregate(self)
 417         rspec =  aggregate.list_resources(version=version, options=options)
 418         return rspec
 419
 420     def describe(self, urns, version, options={}):
 421         aggregate = DummyAggregate(self)
 422         return aggregate.describe(urns, version=version, options=options)
 423
 424     def status (self, urns, options={}):
 425         aggregate = DummyAggregate(self)
 426         desc =  aggregate.describe(urns, version='GENI 3')
 427         status = {'geni_urn': desc['geni_urn'],
 428                   'geni_slivers': desc['geni_slivers']}
 429         return status
 430
 431
 432     def allocate (self, urn, rspec_string, expiration, options={}):
 433         xrn = Xrn(urn)
 434         aggregate = DummyAggregate(self)
 435         slices = DummySlices(self)
 436         slice_record=None
 437         users = options.get('geni_users', [])
 438         if users:
 439             slice_record = users[0].get('slice_record', {})
 440
 441         # parse rspec
 442         rspec = RSpec(rspec_string)
 443         requested_attributes = rspec.version.get_slice_attributes()
 444
 445         # ensure slice record exists
 446         slice = slices.verify_slice(xrn.hrn, slice_record, expiration=expiration, options=options)
 447         # ensure person records exists
 448         #persons = slices.verify_persons(xrn.hrn, slice, users, peer, sfa_peer, options=options)
 449
 450         # add/remove slice from nodes
 451         request_nodes = rspec.version.get_nodes_with_slivers()
 452         nodes = slices.verify_slice_nodes(urn, slice, request_nodes)
 453
 454         return aggregate.describe([xrn.get_urn()], version=rspec.version)
 455
 456     def provision(self, urns, options={}):
 457         # update users
 458         slices = DummySlices(self)
 459         aggregate = DummyAggregate(self)
 460         slivers = aggregate.get_slivers(urns)
 461         slice = slivers[0]
 462         geni_users = options.get('geni_users', [])
 463         #users = slices.verify_users(None, slice, geni_users, options=options)
 464         # update sliver allocation states and set them to geni_provisioned
 465         sliver_ids = [sliver['sliver_id'] for sliver in slivers]
 466         dbsession=self.api.dbsession()
 467         SliverAllocation.set_allocations(sliver_ids, 'geni_provisioned',dbsession)
 468         version_manager = VersionManager()
 469         rspec_version = version_manager.get_version(options['geni_rspec_version'])
 470         return self.describe(urns, rspec_version, options=options)
 471
 472     def delete(self, urns, options={}):
 473         # collect sliver ids so we can update sliver allocation states after
 474         # we remove the slivers.
 475         aggregate = DummyAggregate(self)
 476         slivers = aggregate.get_slivers(urns)
 477         if slivers:
 478             slice_id = slivers[0]['slice_id']
 479             node_ids = []
 480             sliver_ids = []
 481             for sliver in slivers:
 482                 node_ids.append(sliver['node_id'])
 483                 sliver_ids.append(sliver['sliver_id'])
 484
 485             # determine if this is a peer slice
 486             # xxx I wonder if this would not need to use PlSlices.get_peer instead
 487             # in which case plc.peers could be deprecated as this here
 488             # is the only/last call to this last method in plc.peers
 489             slice_hrn = DummyXrn(auth=self.hrn, slicename=slivers[0]['slice_name']).get_hrn()
 490             try:
 491                 self.shell.DeleteSliceFromNodes({'slice_id': slice_id, 'node_ids': node_ids})
 492                 # delete sliver allocation states
 493                 dbsession=self.api.dbsession()
 494                 SliverAllocation.delete_allocations(sliver_ids,dbsession)
 495             finally:
 496                 pass
 497
 498         # prepare return struct
 499         geni_slivers = []
 500         for sliver in slivers:
 501             geni_slivers.append(
 502                 {'geni_sliver_urn': sliver['sliver_id'],
 503                  'geni_allocation_status': 'geni_unallocated',
 504                  'geni_expires': datetime_to_string(utcparse(sliver['expires']))})
 505         return geni_slivers
 506
 507     def renew (self, urns, expiration_time, options={}):
 508         aggregate = DummyAggregate(self)
 509         slivers = aggregate.get_slivers(urns)
 510         if not slivers:
 511             raise SearchFailed(urns)
 512         slice = slivers[0]
 513         requested_time = utcparse(expiration_time)
 514         record = {'expires': int(datetime_to_epoch(requested_time))}
 515         self.shell.UpdateSlice({'slice_id': slice['slice_id'], 'fileds': record})
 516         description = self.describe(urns, 'GENI 3', options)
 517         return description['geni_slivers']
 518
 519     def perform_operational_action (self, urns, action, options={}):
 520         # Dummy doesn't support operational actions. Lets pretend like it
 521         # supports start, but reject everything else.
 522         action = action.lower()
 523         if action not in ['geni_start']:
 524             raise UnsupportedOperation(action)
 525
 526         # fault if sliver is not full allocated (operational status is geni_pending_allocation)
 527         description = self.describe(urns, 'GENI 3', options)
 528         for sliver in description['geni_slivers']:
 529             if sliver['geni_operational_status'] == 'geni_pending_allocation':
 530                 raise UnsupportedOperation(action, "Sliver must be fully allocated (operational status is not geni_pending_allocation)")
 531         #
 532         # Perform Operational Action Here
 533         #
 534
 535         geni_slivers = self.describe(urns, 'GENI 3', options)['geni_slivers']
 536         return geni_slivers
 537
 538     def shutdown (self, xrn, options={}):
 539         xrn = DummyXrn(xrn=xrn, type='slice')
 540         slicename = xrn.pl_slicename()
 541         slices = self.shell.GetSlices({'name': slicename}, ['slice_id'])
 542         if not slices:
 543             raise RecordNotFound(slice_hrn)
 544         slice_id = slices[0]['slice_id']
 545         slice_tags = self.shell.GetSliceTags({'slice_id': slice_id, 'tagname': 'enabled'})
 546         if not slice_tags:
 547             self.shell.AddSliceTag(slice_id, 'enabled', '0')
 548         elif slice_tags[0]['value'] != "0":
 549             tag_id = slice_tags[0]['slice_tag_id']
 550             self.shell.UpdateSliceTag(tag_id, '0')
 551         return 1