sfa/senslab/slabdriver.py

   1 import sys
   2 import subprocess
   3
   4 from datetime import datetime
   5 from dateutil import tz
   6 from time import strftime,gmtime
   7
   8 from sfa.util.faults import MissingSfaInfo , SliverDoesNotExist
   9 from sfa.util.sfalogging import logger
  10 from sfa.util.defaultdict import defaultdict
  11
  12 from sfa.storage.record import Record
  13 from sfa.storage.alchemy import dbsession
  14 from sfa.storage.model import RegRecord
  15
  16
  17 from sfa.trust.certificate import *
  18 from sfa.trust.credential import *
  19 from sfa.trust.gid import GID
  20
  21 from sfa.managers.driver import Driver
  22 from sfa.rspecs.version_manager import VersionManager
  23 from sfa.rspecs.rspec import RSpec
  24
  25 from sfa.util.xrn import hrn_to_urn, urn_to_sliver_id
  26 from sfa.util.plxrn import slicename_to_hrn, hostname_to_hrn, hrn_to_pl_slicename
  27
  28 ## thierry: everything that is API-related (i.e. handling incoming requests)
  29 # is taken care of
  30 # SlabDriver should be really only about talking to the senslab testbed
  31
  32 ## thierry : please avoid wildcard imports :)
  33 from sfa.senslab.OARrestapi import  OARrestapi
  34 from sfa.senslab.LDAPapi import LDAPapi
  35
  36 from sfa.senslab.parsing import parse_filter
  37 from sfa.senslab.slabpostgres import SlabDB, slab_dbsession,SliceSenslab
  38 from sfa.senslab.slabaggregate import SlabAggregate
  39 from sfa.senslab.slabslices import SlabSlices
  40
  41 def list_to_dict(recs, key):
  42     """
  43     convert a list of dictionaries into a dictionary keyed on the
  44     specified dictionary key
  45     """
  46
  47     keys = [rec[key] for rec in recs]
  48     return dict(zip(keys, recs))
  49
  50 # thierry : note
  51 # this inheritance scheme is so that the driver object can receive
  52 # GetNodes or GetSites sorts of calls directly
  53 # and thus minimize the differences in the managers with the pl version
  54 class SlabDriver(Driver):
  55
  56     def __init__(self, config):
  57         Driver.__init__ (self, config)
  58         self.config=config
  59         self.hrn = config.SFA_INTERFACE_HRN
  60
  61         self.root_auth = config.SFA_REGISTRY_ROOT_AUTH
  62
  63
  64         print >>sys.stderr, "\r\n_____________ SFA SENSLAB DRIVER \r\n"
  65
  66         self.oar = OARrestapi()
  67         self.ldap = LDAPapi()
  68         self.time_format = "%Y-%m-%d %H:%M:%S"
  69         self.db = SlabDB(config)
  70         self.cache=None
  71
  72
  73     def sliver_status(self,slice_urn,slice_hrn):
  74         # receive a status request for slice named urn/hrn urn:publicid:IDN+senslab+nturro_slice hrn senslab.nturro_slice
  75         # shall return a structure as described in
  76         # http://groups.geni.net/geni/wiki/GAPI_AM_API_V2#SliverStatus
  77         # NT : not sure if we should implement this or not, but used by sface.
  78
  79
  80         sl = self.GetSlices(slice_filter= slice_hrn, filter_type = 'slice_hrn')
  81         if len(sl) is 0:
  82             raise SliverDoesNotExist("%s  slice_hrn" % (slice_hrn))
  83
  84         print >>sys.stderr, "\r\n \r\n_____________ Sliver status urn %s hrn %s sl %s \r\n " %(slice_urn,slice_hrn,sl)
  85         if sl['oar_job_id'] is not -1:
  86
  87             # report about the local nodes only
  88             nodes_all = self.GetNodes({'hostname':sl['node_ids']},
  89                             ['node_id', 'hostname','site','boot_state'])
  90             nodeall_byhostname = dict([(n['hostname'], n) for n in nodes_all])
  91             nodes = sl['node_ids']
  92             if len(nodes) is 0:
  93                 raise SliverDoesNotExist("No slivers allocated ")
  94
  95
  96             result = {}
  97             top_level_status = 'unknown'
  98             if nodes:
  99                 top_level_status = 'ready'
 100             result['geni_urn'] = slice_urn
 101             result['pl_login'] = sl['job_user']
 102             #result['slab_login'] = sl['job_user']
 103
 104             timestamp = float(sl['startTime']) + float(sl['walltime'])
 105             result['pl_expires'] = strftime(self.time_format, gmtime(float(timestamp)))
 106             #result['slab_expires'] = strftime(self.time_format, gmtime(float(timestamp)))
 107
 108             resources = []
 109             for node in nodes:
 110                 res = {}
 111                 #res['slab_hostname'] = node['hostname']
 112                 #res['slab_boot_state'] = node['boot_state']
 113
 114                 res['pl_hostname'] = nodeall_byhostname[node]['hostname']
 115                 res['pl_boot_state'] = nodeall_byhostname[node]['boot_state']
 116                 res['pl_last_contact'] = strftime(self.time_format, gmtime(float(timestamp)))
 117                 sliver_id = urn_to_sliver_id(slice_urn, sl['record_id_slice'],nodeall_byhostname[node]['node_id'] )
 118                 res['geni_urn'] = sliver_id
 119                 if nodeall_byhostname[node]['boot_state'] == 'Alive':
 120
 121                     res['geni_status'] = 'ready'
 122                 else:
 123                     res['geni_status'] = 'failed'
 124                     top_level_status = 'failed'
 125
 126                 res['geni_error'] = ''
 127
 128                 resources.append(res)
 129
 130             result['geni_status'] = top_level_status
 131             result['geni_resources'] = resources
 132             print >>sys.stderr, "\r\n \r\n_____________ Sliver status resources %s res %s \r\n " %(resources,res)
 133             return result
 134
 135
 136     def create_sliver (self, slice_urn, slice_hrn, creds, rspec_string, users, options):
 137         print>>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver "
 138         aggregate = SlabAggregate(self)
 139
 140         slices = SlabSlices(self)
 141         peer = slices.get_peer(slice_hrn)
 142         sfa_peer = slices.get_sfa_peer(slice_hrn)
 143         slice_record=None
 144
 145         if not isinstance(creds, list):
 146             creds = [creds]
 147
 148         if users:
 149             slice_record = users[0].get('slice_record', {})
 150
 151         # parse rspec
 152         rspec = RSpec(rspec_string)
 153         print>>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver  ============================rspec.version %s " %(rspec.version)
 154
 155
 156         # ensure site record exists?
 157         # ensure slice record exists
 158         slice = slices.verify_slice(slice_hrn, slice_record, peer, sfa_peer, options=options)
 159         requested_attributes = rspec.version.get_slice_attributes()
 160
 161         if requested_attributes:
 162             for attrib_dict in requested_attributes:
 163                 if 'timeslot' in attrib_dict and attrib_dict['timeslot'] is not None:
 164                     slice.update({'timeslot':attrib_dict['timeslot']})
 165         print >>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver  ..... slice %s " %(slice)
 166         # ensure person records exists
 167         persons = slices.verify_persons(slice_hrn, slice, users, peer, sfa_peer, options=options)
 168         # ensure slice attributes exists?
 169
 170
 171         # add/remove slice from nodes
 172         print >>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver  ..... "
 173
 174         requested_slivers = [node.get('component_name') for node in rspec.version.get_nodes_with_slivers()]
 175         print >>sys.stderr, "\r\n \r\n \t=============================== ........... requested_slivers ============================requested_slivers %s " %(requested_slivers)
 176         nodes = slices.verify_slice_nodes(slice, requested_slivers, peer)
 177
 178
 179         return aggregate.get_rspec(slice_xrn=slice_urn, version=rspec.version)
 180
 181
 182     def delete_sliver (self, slice_urn, slice_hrn, creds, options):
 183
 184         slice = self.GetSlices(slice_filter= slice_hrn, filter_type = 'slice_hrn')
 185         print>>sys.stderr, "\r\n \r\n \t\t  SLABDRIVER.PY delete_sliver slice %s" %(slice)
 186         if not slice:
 187             return 1
 188
 189         slices = SlabSlices(self)
 190         # determine if this is a peer slice
 191         # xxx I wonder if this would not need to use PlSlices.get_peer instead
 192         # in which case plc.peers could be deprecated as this here
 193         # is the only/last call to this last method in plc.peers
 194         peer = slices.get_peer(slice_hrn)
 195         try:
 196             if peer:
 197                 self.UnBindObjectFromPeer('slice', slice['record_id_slice'], peer)
 198             self.DeleteSliceFromNodes(slice)
 199         finally:
 200             if peer:
 201                 self.BindObjectToPeer('slice', slice['slice_id'], peer, slice['peer_slice_id'])
 202         return 1
 203
 204
 205     # first 2 args are None in case of resource discovery
 206     def list_resources (self, slice_urn, slice_hrn, creds, options):
 207         #cached_requested = options.get('cached', True)
 208
 209         version_manager = VersionManager()
 210         # get the rspec's return format from options
 211         rspec_version = version_manager.get_version(options.get('geni_rspec_version'))
 212         version_string = "rspec_%s" % (rspec_version)
 213
 214         #panos adding the info option to the caching key (can be improved)
 215         if options.get('info'):
 216             version_string = version_string + "_"+options.get('info', 'default')
 217
 218         # look in cache first
 219         #if cached_requested and self.cache and not slice_hrn:
 220             #rspec = self.cache.get(version_string)
 221             #if rspec:
 222                 #logger.debug("SlabDriver.ListResources: returning cached advertisement")
 223                 #return rspec
 224
 225         #panos: passing user-defined options
 226
 227         aggregate = SlabAggregate(self)
 228         origin_hrn = Credential(string=creds[0]).get_gid_caller().get_hrn()
 229         #print>>sys.stderr, " \r\n \r\n \t SLABDRIVER list_resources origin_hrn %s" %(origin_hrn)
 230         options.update({'origin_hrn':origin_hrn})
 231         #print>>sys.stderr, " \r\n \r\n \t SLABDRIVER  list_resources options %s" %(options)
 232         rspec =  aggregate.get_rspec(slice_xrn=slice_urn, version=rspec_version,
 233                                      options=options)
 234         print>>sys.stderr, " \r\n \r\n \t SLABDRIVER list_resources rspec "
 235         # cache the result
 236         #if self.cache and not slice_hrn:
 237             #logger.debug("Slab.ListResources: stores advertisement in cache")
 238             #self.cache.add(version_string, rspec)
 239
 240         return rspec
 241
 242
 243     def list_slices (self, creds, options):
 244         # look in cache first
 245         #if self.cache:
 246             #slices = self.cache.get('slices')
 247             #if slices:
 248                 #logger.debug("PlDriver.list_slices returns from cache")
 249                 #return slices
 250
 251         # get data from db
 252         print>>sys.stderr, " \r\n \t\t SLABDRIVER.PY list_slices"
 253         slices = self.GetSlices()
 254         slice_hrns = [slicename_to_hrn(self.hrn, slice['slice_hrn']) for slice in slices]
 255         slice_urns = [hrn_to_urn(slice_hrn, 'slice') for slice_hrn in slice_hrns]
 256
 257         # cache the result
 258         #if self.cache:
 259             #logger.debug ("SlabDriver.list_slices stores value in cache")
 260             #self.cache.add('slices', slice_urns)
 261
 262         return slice_urns
 263
 264     #No site or node register supported
 265     def register (self, sfa_record, hrn, pub_key):
 266         type = sfa_record['type']
 267         slab_record = self.sfa_fields_to_slab_fields(type, hrn, sfa_record)
 268
 269
 270         if type == 'slice':
 271             acceptable_fields=['url', 'instantiation', 'name', 'description']
 272             for key in slab_record.keys():
 273                 if key not in acceptable_fields:
 274                     slab_record.pop(key)
 275             print>>sys.stderr, " \r\n \t\t SLABDRIVER.PY register"
 276             slices = self.GetSlices(slice_filter =slab_record['hrn'], filter_type = 'slice_hrn')
 277             if not slices:
 278                     pointer = self.AddSlice(slab_record)
 279             else:
 280                     pointer = slices[0]['slice_id']
 281
 282         elif type == 'user':
 283             persons = self.GetPersons([sfa_record['hrn']])
 284             if not persons:
 285                 pointer = self.AddPerson(dict(sfa_record))
 286                 #add in LDAP
 287             else:
 288                 pointer = persons[0]['person_id']
 289
 290             #Does this make sense to senslab ?
 291             #if 'enabled' in sfa_record and sfa_record['enabled']:
 292                 #self.UpdatePerson(pointer, {'enabled': sfa_record['enabled']})
 293
 294             # add this person to the site only if she is being added for the first
 295             # time by sfa and doesont already exist in plc
 296             if not persons or not persons[0]['site_ids']:
 297                 login_base = get_leaf(sfa_record['authority'])
 298                 self.AddPersonToSite(pointer, login_base)
 299
 300             # What roles should this user have?
 301             self.AddRoleToPerson('user', pointer)
 302             # Add the user's key
 303             if pub_key:
 304                 self.AddPersonKey(pointer, {'key_type' : 'ssh', 'key' : pub_key})
 305
 306         #No node adding outside OAR
 307
 308         return pointer
 309
 310     #No site or node record update allowed
 311     def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
 312         pointer = old_sfa_record['pointer']
 313         type = old_sfa_record['type']
 314
 315         # new_key implemented for users only
 316         if new_key and type not in [ 'user' ]:
 317             raise UnknownSfaType(type)
 318
 319         #if (type == "authority"):
 320             #self.shell.UpdateSite(pointer, new_sfa_record)
 321
 322         if type == "slice":
 323             slab_record=self.sfa_fields_to_slab_fields(type, hrn, new_sfa_record)
 324             if 'name' in slab_record:
 325                 slab_record.pop('name')
 326                 self.UpdateSlice(pointer, slab_record)
 327
 328         elif type == "user":
 329             update_fields = {}
 330             all_fields = new_sfa_record
 331             for key in all_fields.keys():
 332                 if key in ['first_name', 'last_name', 'title', 'email',
 333                            'password', 'phone', 'url', 'bio', 'accepted_aup',
 334                            'enabled']:
 335                     update_fields[key] = all_fields[key]
 336             self.UpdatePerson(pointer, update_fields)
 337
 338             if new_key:
 339                 # must check this key against the previous one if it exists
 340                 persons = self.GetPersons([pointer], ['key_ids'])
 341                 person = persons[0]
 342                 keys = person['key_ids']
 343                 keys = self.GetKeys(person['key_ids'])
 344
 345                 # Delete all stale keys
 346                 key_exists = False
 347                 for key in keys:
 348                     if new_key != key['key']:
 349                         self.DeleteKey(key['key_id'])
 350                     else:
 351                         key_exists = True
 352                 if not key_exists:
 353                     self.AddPersonKey(pointer, {'key_type': 'ssh', 'key': new_key})
 354
 355
 356         return True
 357
 358
 359     def remove (self, sfa_record):
 360         type=sfa_record['type']
 361         hrn=sfa_record['hrn']
 362         record_id= sfa_record['record_id']
 363         if type == 'user':
 364             username = hrn.split(".")[len(hrn.split(".")) -1]
 365             #get user in ldap
 366             persons = self.GetPersons(username)
 367             # only delete this person if he has site ids. if he doesnt, it probably means
 368             # he was just removed from a site, not actually deleted
 369             if persons and persons[0]['site_ids']:
 370                 self.DeletePerson(username)
 371         elif type == 'slice':
 372             if self.GetSlices(slice_filter = hrn, filter_type = 'slice_hrn'):
 373                 self.DeleteSlice(hrn)
 374
 375         #elif type == 'authority':
 376             #if self.GetSites(pointer):
 377                 #self.DeleteSite(pointer)
 378
 379         return True
 380
 381     def GetPeers (self,auth = None, peer_filter=None, return_fields=None):
 382
 383         existing_records = {}
 384         existing_hrns_by_types= {}
 385         print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers auth = %s, peer_filter %s, return_field %s " %(auth , peer_filter, return_fields)
 386         all_records = dbsession.query(RegRecord).filter(RegRecord.type.like('%authority%')).all()
 387         for record in all_records:
 388             existing_records[record.hrn] = record
 389             if record.type not in existing_hrns_by_types:
 390                 existing_hrns_by_types[record.type] = [record.hrn]
 391                 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers \t NOT IN existing_hrns_by_types %s " %( existing_hrns_by_types)
 392             else:
 393
 394                 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers \t INNN  type %s hrn %s " %( record.type,record.hrn )
 395                 existing_hrns_by_types[record.type].append(record.hrn)
 396                 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers \t INNN existing_hrns_by_types %s " %( existing_hrns_by_types)
 397                 #existing_hrns_by_types.update({record.type:(existing_hrns_by_types[record.type].append(record.hrn))})
 398
 399         print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers        existing_hrns_by_types %s " %( existing_hrns_by_types)
 400         records_list= []
 401
 402         try:
 403             for hrn in existing_hrns_by_types['authority+sa']:
 404                 records_list.append(existing_records[hrn])
 405                 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers  records_list  %s " %(records_list)
 406
 407         except:
 408                 pass
 409
 410         if not peer_filter and not return_fields:
 411             return records_list
 412         return_records = parse_filter(records_list,peer_filter, 'peers', return_fields)
 413
 414         return return_records
 415
 416
 417
 418     def GetPersons(self, person_filter=None, return_fields=None):
 419
 420         person_list = self.ldap.ldapFind({'authority': self.root_auth })
 421
 422         #check = False
 423         #if person_filter and isinstance(person_filter, dict):
 424             #for k in  person_filter.keys():
 425                 #if k in person_list[0].keys():
 426                     #check = True
 427
 428         return_person_list = parse_filter(person_list,person_filter ,'persons', return_fields)
 429         if return_person_list:
 430             print>>sys.stderr, " \r\n GetPersons person_filter %s return_fields %s  " %(person_filter,return_fields)
 431             return return_person_list
 432
 433     def GetTimezone(self):
 434         server_timestamp,server_tz = self.oar.parser.SendRequest("GET_timezone")
 435         return server_timestamp,server_tz
 436
 437
 438     def DeleteJobs(self, job_id, slice_hrn):
 439         if not job_id:
 440             return
 441         username  = slice_hrn.split(".")[-1].rstrip("_slice")
 442         reqdict = {}
 443         reqdict['method'] = "delete"
 444         reqdict['strval'] = str(job_id)
 445         answer = self.oar.POSTRequestToOARRestAPI('DELETE_jobs_id',reqdict,username)
 446         print>>sys.stderr, "\r\n \r\n  jobid  DeleteJobs %s "  %(answer)
 447
 448
 449     def GetJobs(self,job_id= None, resources=True,return_fields=None, username = None):
 450         #job_resources=['reserved_resources', 'assigned_resources','job_id', 'job_uri', 'assigned_nodes',\
 451         #'api_timestamp']
 452         #assigned_res = ['resource_id', 'resource_uri']
 453         #assigned_n = ['node', 'node_uri']
 454
 455         if job_id and resources is False:
 456             req = "GET_jobs_id"
 457             node_list_k = 'assigned_network_address'
 458
 459         if job_id and resources :
 460             req = "GET_jobs_id_resources"
 461             node_list_k = 'reserved_resources'
 462
 463         #Get job info from OAR
 464         job_info = self.oar.parser.SendRequest(req, job_id, username)
 465         print>>sys.stderr, "\r\n \r\n \t\t GetJobs  %s " %(job_info)
 466
 467         if 'state' in job_info :
 468             if job_info['state'] == 'Terminated':
 469                 print>>sys.stderr, "\r\n \r\n \t\t GetJobs TERMINELEBOUSIN "
 470                 return None
 471             if job_info['state'] == 'Error':
 472                 print>>sys.stderr, "\r\n \r\n \t\t GetJobs ERROR message %s " %(job_info)
 473                 return None
 474
 475         #Get a dict of nodes . Key :hostname of the node
 476         node_list = self.GetNodes()
 477         node_hostname_list = []
 478         for node in node_list:
 479             node_hostname_list.append(node['hostname'])
 480         node_dict = dict(zip(node_hostname_list,node_list))
 481         try :
 482             liste =job_info[node_list_k]
 483             #print>>sys.stderr, "\r\n \r\n \t\t GetJobs resources  job_info liste%s" %(liste)
 484             for k in range(len(liste)):
 485                job_info[node_list_k][k] = node_dict[job_info[node_list_k][k]]['hostname']
 486
 487             #print>>sys.stderr, "\r\n \r\n \t\t YYYYYYYYYYYYGetJobs resources  job_info %s" %(job_info)
 488             #Replaces the previous entry "assigned_network_address" / "reserved_resources"
 489             #with "node_ids"
 490             job_info.update({'node_ids':job_info[node_list_k]})
 491             del job_info[node_list_k]
 492             return job_info
 493
 494         except KeyError:
 495             print>>sys.stderr, "\r\n \r\n \t\t GetJobs KEYERROR "
 496
 497     def GetReservedNodes(self):
 498         # this function returns a list of all the nodes already involved in an oar job
 499
 500        jobs=self.oar.parser.SendRequest("GET_jobs_details")
 501        nodes=[]
 502        for j in jobs :
 503           nodes=j['assigned_network_address']+nodes
 504        return nodes
 505
 506     def GetNodes(self,node_filter= None, return_fields=None):
 507         node_dict =self.oar.parser.SendRequest("GET_resources_full")
 508
 509         return_node_list = []
 510         if not (node_filter or return_fields):
 511                 return_node_list = node_dict.values()
 512                 return return_node_list
 513
 514         return_node_list= parse_filter(node_dict.values(),node_filter ,'node', return_fields)
 515         return return_node_list
 516
 517
 518     def GetSites(self, site_filter = None, return_fields=None):
 519         site_dict =self.oar.parser.SendRequest("GET_sites")
 520         return_site_list = []
 521         if not ( site_filter or return_fields):
 522                 return_site_list = site_dict.values()
 523                 return return_site_list
 524
 525         return_site_list = parse_filter(site_dict.values(), site_filter,'site', return_fields)
 526         return return_site_list
 527
 528
 529     def GetSlices(self,slice_filter = None, filter_type = None, return_fields=None):
 530         return_slice_list = []
 531         slicerec  = {}
 532         rec = {}
 533         ftypes = ['slice_hrn', 'record_id_user']
 534         if filter_type and filter_type in ftypes:
 535             if filter_type == 'slice_hrn':
 536                 slicerec = slab_dbsession.query(SliceSenslab).filter_by(slice_hrn = slice_filter).first()
 537             if filter_type == 'record_id_user':
 538                 slicerec = slab_dbsession.query(SliceSenslab).filter_by(record_id_user = slice_filter).first()
 539
 540             if slicerec:
 541                 rec = slicerec.dumpquerytodict()
 542                 login = slicerec.slice_hrn.split(".")[1].split("_")[0]
 543                 #print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY slicerec GetSlices   %s " %(slicerec)
 544                 if slicerec.oar_job_id is not -1:
 545                     rslt = self.GetJobs( slicerec.oar_job_id, resources=False, username = login )
 546                     #print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY  GetSlices  GetJobs  %s " %(rslt)
 547                     if rslt :
 548                         rec.update(rslt)
 549                         rec.update({'hrn':str(rec['slice_hrn'])})
 550                         #If GetJobs is empty, this means the job is now in the 'Terminated' state
 551                         #Update the slice record
 552                     else :
 553                         self.db.update_job(slice_filter, job_id = -1)
 554                         rec['oar_job_id'] = -1
 555                         rec.update({'hrn':str(rec['slice_hrn'])})
 556
 557                 try:
 558                     rec['node_ids'] = rec['node_list']
 559                 except KeyError:
 560                     pass
 561
 562                 #print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY  GetSlices  rec  %s" %(rec)
 563
 564             return rec
 565
 566
 567         else:
 568             return_slice_list = slab_dbsession.query(SliceSenslab).all()
 569
 570         print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY  GetSlices  slices %s slice_filter %s " %(return_slice_list,slice_filter)
 571
 572         #if return_fields:
 573             #return_slice_list  = parse_filter(sliceslist, slice_filter,'slice', return_fields)
 574
 575
 576
 577         return return_slice_list
 578
 579
 580
 581
 582     def testbed_name (self): return "senslab2"
 583
 584     # 'geni_request_rspec_versions' and 'geni_ad_rspec_versions' are mandatory
 585     def aggregate_version (self):
 586         version_manager = VersionManager()
 587         ad_rspec_versions = []
 588         request_rspec_versions = []
 589         for rspec_version in version_manager.versions:
 590             if rspec_version.content_type in ['*', 'ad']:
 591                 ad_rspec_versions.append(rspec_version.to_dict())
 592             if rspec_version.content_type in ['*', 'request']:
 593                 request_rspec_versions.append(rspec_version.to_dict())
 594         return {
 595             'testbed':self.testbed_name(),
 596             'geni_request_rspec_versions': request_rspec_versions,
 597             'geni_ad_rspec_versions': ad_rspec_versions,
 598             }
 599
 600
 601
 602
 603
 604
 605     ##
 606     # Convert SFA fields to PLC fields for use when registering up updating
 607     # registry record in the PLC database
 608     #
 609     # @param type type of record (user, slice, ...)
 610     # @param hrn human readable name
 611     # @param sfa_fields dictionary of SFA fields
 612     # @param slab_fields dictionary of PLC fields (output)
 613
 614     def sfa_fields_to_slab_fields(self, type, hrn, record):
 615
 616         def convert_ints(tmpdict, int_fields):
 617             for field in int_fields:
 618                 if field in tmpdict:
 619                     tmpdict[field] = int(tmpdict[field])
 620
 621         slab_record = {}
 622         #for field in record:
 623         #    slab_record[field] = record[field]
 624
 625         if type == "slice":
 626             #instantion used in get_slivers ?
 627             if not "instantiation" in slab_record:
 628                 slab_record["instantiation"] = "senslab-instantiated"
 629             slab_record["hrn"] = hrn_to_pl_slicename(hrn)
 630             print >>sys.stderr, "\r\n \r\n \t SLABDRIVER.PY sfa_fields_to_slab_fields slab_record %s hrn_to_pl_slicename(hrn) hrn %s " %(slab_record['hrn'], hrn)
 631             if "url" in record:
 632                slab_record["url"] = record["url"]
 633             if "description" in record:
 634                 slab_record["description"] = record["description"]
 635             if "expires" in record:
 636                 slab_record["expires"] = int(record["expires"])
 637
 638         #nodes added by OAR only and then imported to SFA
 639         #elif type == "node":
 640             #if not "hostname" in slab_record:
 641                 #if not "hostname" in record:
 642                     #raise MissingSfaInfo("hostname")
 643                 #slab_record["hostname"] = record["hostname"]
 644             #if not "model" in slab_record:
 645                 #slab_record["model"] = "geni"
 646
 647         #One authority only
 648         #elif type == "authority":
 649             #slab_record["login_base"] = hrn_to_slab_login_base(hrn)
 650
 651             #if not "name" in slab_record:
 652                 #slab_record["name"] = hrn
 653
 654             #if not "abbreviated_name" in slab_record:
 655                 #slab_record["abbreviated_name"] = hrn
 656
 657             #if not "enabled" in slab_record:
 658                 #slab_record["enabled"] = True
 659
 660             #if not "is_public" in slab_record:
 661                 #slab_record["is_public"] = True
 662
 663         return slab_record
 664
 665
 666     def LaunchExperimentOnOAR(self,  slice_dict, added_nodes, slice_user=None):
 667
 668         site_list = []
 669         nodeid_list =[]
 670         resource = ""
 671         reqdict = {}
 672         slice_name = slice_dict['name']
 673         try:
 674             slot = slice_dict['timeslot']
 675             print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR slot %s   " %(slot)
 676         except KeyError:
 677             #Running on default parameters
 678             #XP immediate , 10 mins
 679             slot = {'date':None,'start_time':None, 'timezone':None,'duration':None }#10 min
 680
 681
 682         reqdict['property'] ="network_address in ("
 683         for node in added_nodes:
 684             #Get the ID of the node : remove the root auth and put the site in a separate list
 685             s=node.split(".")
 686             # NT: it's not clear for me if the nodenames will have the senslab prefix
 687             # so lets take the last part only, for now.
 688             lastpart=s[-1]
 689             #if s[0] == self.root_auth :
 690             # Again here it's not clear if nodes will be prefixed with <site>_, lets split and tanke the last part for now.
 691             s=lastpart.split("_")
 692             nodeid=s[-1]
 693             reqdict['property'] += "'"+ nodeid +"', "
 694             nodeid_list.append(nodeid)
 695             #site_list.append( l[0] )
 696
 697
 698         reqdict['property'] =  reqdict['property'][0: len( reqdict['property'])-2] +")"
 699         reqdict['resource'] ="network_address="+ str(len(nodeid_list))
 700
 701         if slot['duration']:
 702             walltime = slot['duration'].split(":")
 703             # Fixing the walltime by adding a few delays. First put the walltime in seconds
 704             # oarAdditionalDelay = 20; additional delay for /bin/sleep command to
 705             # take in account  prologue and epilogue scripts execution
 706             # int walltimeAdditionalDelay = 120;  additional delay
 707
 708             desired_walltime =  int(walltime[0])*3600 + int(walltime[1]) * 60 + int(walltime[2])
 709             total_walltime = desired_walltime + 140 #+2 min 20
 710             sleep_walltime = desired_walltime + 20 #+20 sec
 711             print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR desired_walltime %s  total_walltime %s sleep_walltime %s  " %(desired_walltime,total_walltime,sleep_walltime)
 712             #Put the walltime back in str form
 713             #First get the hours
 714             walltime[0] = str(total_walltime / 3600)
 715             total_walltime = total_walltime - 3600 * int(walltime[0])
 716             #Get the remaining minutes
 717             walltime[1] = str(total_walltime / 60)
 718             total_walltime =  total_walltime - 60 * int(walltime[1])
 719             #Get the seconds
 720             walltime[2] = str(total_walltime)
 721             print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR  walltime %s " %(walltime)
 722
 723             reqdict['resource']+= ",walltime=" + str(walltime[0]) + ":" + str(walltime[1]) + ":" + str(walltime[2])
 724             reqdict['script_path'] = "/bin/sleep " + str(sleep_walltime)
 725         else:
 726             reqdict['resource']+= ",walltime=" + str(00) + ":" + str(12) + ":" + str(20) #+2 min 20
 727             reqdict['script_path'] = "/bin/sleep 620" #+20 sec
 728         #In case of a scheduled experiment (not immediate)
 729         #To run an XP immediately, don't specify date and time in RSpec
 730         #They will be set to None.
 731         if slot['date'] and slot['start_time']:
 732             if slot['timezone'] is '' or slot['timezone'] is None:
 733                 #assume it is server timezone
 734                 server_timestamp,server_tz = self.GetTimezone()
 735                 from_zone=tz.gettz(server_tz)
 736                 print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR  timezone not specified  server_tz %s from_zone  %s" %(server_tz,from_zone)
 737             else:
 738                 #Get zone of the user from the reservation time given in the rspec
 739                 from_zone = tz.gettz(slot['timezone'])
 740
 741             date = str(slot['date'])  + " " + str(slot['start_time'])
 742             user_datetime = datetime.datetime.strptime(date, self.time_format)
 743             user_datetime = user_datetime.replace(tzinfo = from_zone)
 744
 745             #Convert to UTC zone
 746             to_zone = tz.tzutc()
 747             utc_date = user_datetime.astimezone(to_zone)
 748             #Readable time accpeted by OAR
 749             reqdict['reservation']= utc_date.strftime(self.time_format)
 750
 751             print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR  reqdict['reservation'] %s " %(reqdict['reservation'])
 752
 753         else:
 754             # Immediate XP
 755             # reservations are performed in the oar server timebase, so :
 756             # 1- we get the server time(in UTC tz )/server timezone
 757             # 2- convert the server UTC time in its timezone
 758             # 3- add a custom delay to this time
 759             # 4- convert this time to a readable form and it for the reservation request.
 760             server_timestamp,server_tz = self.GetTimezone()
 761             s_tz=tz.gettz(server_tz)
 762             UTC_zone = tz.gettz("UTC")
 763             #weird... datetime.fromtimestamp should work since we do from datetime import datetime
 764             utc_server= datetime.datetime.fromtimestamp(float(server_timestamp)+20,UTC_zone)
 765             server_localtime=utc_server.astimezone(s_tz)
 766
 767             print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR server_timestamp %s server_tz %s slice_name %s added_nodes %s username %s reqdict %s " %(server_timestamp,server_tz,slice_name,added_nodes,slice_user, reqdict )
 768             readable_time = server_localtime.strftime(self.time_format)
 769
 770             print >>sys.stderr,"  \r\n \r\n \t\t\t\tAPRES ParseTimezone readable_time %s timestanp %s  " %(readable_time ,server_timestamp)
 771             reqdict['reservation'] = readable_time
 772
 773
 774         reqdict['type'] = "deploy"
 775         reqdict['directory']= ""
 776         reqdict['name']= "TestSandrine"
 777
 778
 779         # first step : start the OAR job and update the job
 780         print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR reqdict   %s \r\n site_list   %s"  %(reqdict,site_list)
 781
 782         answer = self.oar.POSTRequestToOARRestAPI('POST_job',reqdict,slice_user)
 783         print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR jobid   %s "  %(answer)
 784         try:
 785             jobid = answer['id']
 786         except KeyError:
 787              print>>sys.stderr, "\r\n AddSliceTonode Impossible to create job  %s "  %( answer)
 788              return
 789
 790         print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR jobid    %s added_nodes  %s slice_user %s"  %(jobid,added_nodes,slice_user)
 791         self.db.update_job( slice_name, jobid ,added_nodes)
 792
 793
 794         # second step : configure the experiment
 795         # we need to store the nodes in a yaml (well...) file like this :
 796         # [1,56,23,14,45,75] with name /tmp/sfa<jobid>.json
 797         f=open('/tmp/sfa/'+str(jobid)+'.json','w')
 798         f.write('[')
 799         f.write(str(added_nodes[0].strip('node')))
 800         for node in added_nodes[1:len(added_nodes)] :
 801             f.write(','+node.strip('node'))
 802         f.write(']')
 803         f.close()
 804
 805         # third step : call the senslab-experiment wrapper
 806         #command= "java -jar target/sfa-1.0-jar-with-dependencies.jar "+str(jobid)+" "+slice_user
 807         javacmdline="/usr/bin/java"
 808         jarname="/opt/senslabexperimentwrapper/sfa-1.0-jar-with-dependencies.jar"
 809         #ret=subprocess.check_output(["/usr/bin/java", "-jar", ", str(jobid), slice_user])
 810         output = subprocess.Popen([javacmdline, "-jar", jarname, str(jobid), slice_user],stdout=subprocess.PIPE).communicate()[0]
 811
 812         print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR wrapper returns   %s "  %(output)
 813         return
 814
 815
 816     #Delete the jobs and updates the job id in the senslab table
 817     #to set it to -1
 818     #Does not clear the node list
 819     def DeleteSliceFromNodes(self, slice_record):
 820          # Get user information
 821
 822         self.DeleteJobs(slice_record['oar_job_id'], slice_record['hrn'])
 823         self.db.update_job(slice_record['hrn'], job_id = -1)
 824         return
 825
 826
 827
 828     def fill_record_sfa_info(self, records):
 829
 830         def startswith(prefix, values):
 831             return [value for value in values if value.startswith(prefix)]
 832
 833         # get person ids
 834         person_ids = []
 835         site_ids = []
 836         for record in records:
 837             person_ids.extend(record.get("person_ids", []))
 838             site_ids.extend(record.get("site_ids", []))
 839             if 'site_id' in record:
 840                 site_ids.append(record['site_id'])
 841
 842         #print>>sys.stderr, "\r\n \r\n _fill_record_sfa_info ___person_ids %s \r\n \t\t site_ids %s " %(person_ids, site_ids)
 843
 844         # get all pis from the sites we've encountered
 845         # and store them in a dictionary keyed on site_id
 846         site_pis = {}
 847         if site_ids:
 848             pi_filter = {'|roles': ['pi'], '|site_ids': site_ids}
 849             pi_list = self.GetPersons( pi_filter, ['person_id', 'site_ids'])
 850             #print>>sys.stderr, "\r\n \r\n _fill_record_sfa_info ___ GetPersons ['person_id', 'site_ids'] pi_ilist %s" %(pi_list)
 851
 852             for pi in pi_list:
 853                 # we will need the pi's hrns also
 854                 person_ids.append(pi['person_id'])
 855
 856                 # we also need to keep track of the sites these pis
 857                 # belong to
 858                 for site_id in pi['site_ids']:
 859                     if site_id in site_pis:
 860                         site_pis[site_id].append(pi)
 861                     else:
 862                         site_pis[site_id] = [pi]
 863
 864         # get sfa records for all records associated with these records.
 865         # we'll replace pl ids (person_ids) with hrns from the sfa records
 866         # we obtain
 867
 868         # get the sfa records
 869         #table = SfaTable()
 870         existing_records = {}
 871         all_records = dbsession.query(RegRecord).all()
 872         for record in all_records:
 873             existing_records[(record.type,record.pointer)] = record
 874
 875         print >>sys.stderr, " \r\r\n SLABDRIVER fill_record_sfa_info existing_records %s "  %(existing_records)
 876         person_list, persons = [], {}
 877         #person_list = table.find({'type': 'user', 'pointer': person_ids})
 878         try:
 879             for p_id in person_ids:
 880                 person_list.append( existing_records.get(('user',p_id)))
 881         except KeyError:
 882             print >>sys.stderr, " \r\r\n SLABDRIVER fill_record_sfa_info ERRRRRRRRRROR"
 883
 884         # create a hrns keyed on the sfa record's pointer.
 885         # Its possible for  multiple records to have the same pointer so
 886         # the dict's value will be a list of hrns.
 887         persons = defaultdict(list)
 888         for person in person_list:
 889             persons[person['pointer']].append(person)
 890
 891         # get the pl records
 892         slab_person_list, slab_persons = [], {}
 893         slab_person_list = self.GetPersons(person_ids, ['person_id', 'roles'])
 894         slab_persons = list_to_dict(slab_person_list, 'person_id')
 895         #print>>sys.stderr, "\r\n \r\n _fill_record_sfa_info ___  _list %s \r\n \t\t SenslabUsers.GetPersons ['person_id', 'roles'] slab_persons %s \r\n records %s" %(slab_person_list, slab_persons,records)
 896         # fill sfa info
 897
 898         for record in records:
 899             # skip records with no pl info (top level authorities)
 900             #Sandrine 24 oct 11 2 lines
 901             #if record['pointer'] == -1:
 902                 #continue
 903             sfa_info = {}
 904             type = record['type']
 905             if (type == "slice"):
 906                 # all slice users are researchers
 907                 #record['geni_urn'] = hrn_to_urn(record['hrn'], 'slice')  ? besoin ou pas ?
 908                 record['PI'] = []
 909                 record['researcher'] = []
 910                 for person_id in record.get('person_ids', []):
 911                          #Sandrine 24 oct 11 line
 912                 #for person_id in record['person_ids']:
 913                     hrns = [person['hrn'] for person in persons[person_id]]
 914                     record['researcher'].extend(hrns)
 915
 916                 # pis at the slice's site
 917                 slab_pis = site_pis[record['site_id']]
 918                 pi_ids = [pi['person_id'] for pi in slab_pis]
 919                 for person_id in pi_ids:
 920                     hrns = [person['hrn'] for person in persons[person_id]]
 921                     record['PI'].extend(hrns)
 922                 record['geni_urn'] = hrn_to_urn(record['hrn'], 'slice')
 923                 record['geni_creator'] = record['PI']
 924
 925             elif (type == "authority"):
 926                 record['PI'] = []
 927                 record['operator'] = []
 928                 record['owner'] = []
 929                 for pointer in record['person_ids']:
 930                     if pointer not in persons or pointer not in slab_persons:
 931                         # this means there is not sfa or pl record for this user
 932                         continue
 933                     hrns = [person['hrn'] for person in persons[pointer]]
 934                     roles = slab_persons[pointer]['roles']
 935                     if 'pi' in roles:
 936                         record['PI'].extend(hrns)
 937                     if 'tech' in roles:
 938                         record['operator'].extend(hrns)
 939                     if 'admin' in roles:
 940                         record['owner'].extend(hrns)
 941                     # xxx TODO: OrganizationName
 942             elif (type == "node"):
 943                 sfa_info['dns'] = record.get("hostname", "")
 944                 # xxx TODO: URI, LatLong, IP, DNS
 945
 946             elif (type == "user"):
 947                  sfa_info['email'] = record.get("email", "")
 948                  sfa_info['geni_urn'] = hrn_to_urn(record['hrn'], 'user')
 949                  sfa_info['geni_certificate'] = record['gid']
 950                 # xxx TODO: PostalAddress, Phone
 951
 952             #print>>sys.stderr, "\r\n \r\rn \t\t \t <<<<<<<<<<<<<<<<<<<<<<<<  fill_record_sfa_info sfa_info %s  \r\n record %s : "%(sfa_info,record)
 953             record.update(sfa_info)
 954
 955     def augment_records_with_testbed_info (self, sfa_records):
 956         return self.fill_record_info (sfa_records)
 957
 958     def fill_record_info(self, records):
 959         """
 960         Given a SFA record, fill in the senslab specific and SFA specific
 961         fields in the record.
 962         """
 963
 964         print >>sys.stderr, "\r\n \t\t  SLABDRIVER.PY fill_record_info 000000000 fill_record_info %s  " %(records)
 965         if not isinstance(records, list):
 966             records = [records]
 967
 968         parkour = records
 969         try:
 970             for record in parkour:
 971
 972                 if str(record['type']) == 'slice':
 973                     #print >>sys.stderr, "\r\n \t\t  SLABDRIVER.PY  fill_record_info \t \t record %s" %(record)
 974                     #sfatable = SfaTable()
 975
 976                     #existing_records_by_id = {}
 977                     #all_records = dbsession.query(RegRecord).all()
 978                     #for rec in all_records:
 979                         #existing_records_by_id[rec.record_id] = rec
 980                     #print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY  fill_record_info \t\t existing_records_by_id %s" %(existing_records_by_id[record['record_id']])
 981
 982                     #recslice = self.db.find('slice',{'slice_hrn':str(record['hrn'])})
 983                     #recslice = slab_dbsession.query(SliceSenslab).filter_by(slice_hrn = str(record['hrn'])).first()
 984                     recslice = self.GetSlices(slice_filter =  str(record['hrn']), filter_type = 'slice_hrn')
 985                     #print >>sys.stderr, "\r\n \t\t  SLABDRIVER.PY fill_record_info \t\t HOY HOY reclise %s" %(recslice)
 986                     #if isinstance(recslice,list) and len(recslice) == 1:
 987                         #recslice = recslice[0]
 988
 989                     recuser = dbsession.query(RegRecord).filter_by(record_id = recslice['record_id_user']).first()
 990                     #existing_records_by_id[recslice['record_id_user']]
 991                     #print >>sys.stderr, "\r\n \t\t  SLABDRIVER.PY fill_record_info \t\t recuser %s" %(recuser)
 992
 993
 994                     record.update({'PI':[recuser.hrn],
 995                     'researcher': [recuser.hrn],
 996                     'name':record['hrn'],
 997                     'oar_job_id':recslice['oar_job_id'],
 998                     'node_ids': [],
 999                     'person_ids':[recslice['record_id_user']]})
1000
1001                 elif str(record['type']) == 'user':
1002                     print >>sys.stderr, "\r\n \t\t  SLABDRIVER.PY fill_record_info USEEEEEEEEEERDESU!"
1003
1004                     rec = self.GetSlices(slice_filter = record['record_id'], filter_type = 'record_id_user')
1005                     #Append record in records list, therfore fetches user and slice info again(one more loop)
1006                     #Will update PIs and researcher for the slice
1007
1008                     rec.update({'type':'slice','hrn':rec['slice_hrn']})
1009                     records.append(rec)
1010                     #print >>sys.stderr, "\r\n \t\t  SLABDRIVER.PY fill_record_info ADDING SLIC EINFO rec %s" %(rec)
1011
1012             print >>sys.stderr, "\r\n \t\t  SLABDRIVER.PY fill_record_info OKrecords %s" %(records)
1013         except TypeError:
1014             print >>sys.stderr, "\r\n \t\t SLABDRIVER fill_record_info  EXCEPTION RECORDS : %s" %(records)
1015             return
1016
1017         #self.fill_record_slab_info(records)
1018         ##print >>sys.stderr, "\r\n \t\t after fill_record_slab_info %s" %(records)
1019         #self.fill_record_sfa_info(records)
1020         #print >>sys.stderr, "\r\n \t\t after fill_record_sfa_info"
1021
1022
1023
1024
1025
1026     #def update_membership_list(self, oldRecord, record, listName, addFunc, delFunc):
1027         ## get a list of the HRNs tht are members of the old and new records
1028         #if oldRecord:
1029             #oldList = oldRecord.get(listName, [])
1030         #else:
1031             #oldList = []
1032         #newList = record.get(listName, [])
1033
1034         ## if the lists are the same, then we don't have to update anything
1035         #if (oldList == newList):
1036             #return
1037
1038         ## build a list of the new person ids, by looking up each person to get
1039         ## their pointer
1040         #newIdList = []
1041         #table = SfaTable()
1042         #records = table.find({'type': 'user', 'hrn': newList})
1043         #for rec in records:
1044             #newIdList.append(rec['pointer'])
1045
1046         ## build a list of the old person ids from the person_ids field
1047         #if oldRecord:
1048             #oldIdList = oldRecord.get("person_ids", [])
1049             #containerId = oldRecord.get_pointer()
1050         #else:
1051             ## if oldRecord==None, then we are doing a Register, instead of an
1052             ## update.
1053             #oldIdList = []
1054             #containerId = record.get_pointer()
1055
1056     ## add people who are in the new list, but not the oldList
1057         #for personId in newIdList:
1058             #if not (personId in oldIdList):
1059                 #addFunc(self.plauth, personId, containerId)
1060
1061         ## remove people who are in the old list, but not the new list
1062         #for personId in oldIdList:
1063             #if not (personId in newIdList):
1064                 #delFunc(self.plauth, personId, containerId)
1065
1066     #def update_membership(self, oldRecord, record):
1067         #print >>sys.stderr, " \r\n \r\n ***SLABDRIVER.PY update_membership record ", record
1068         #if record.type == "slice":
1069             #self.update_membership_list(oldRecord, record, 'researcher',
1070                                         #self.users.AddPersonToSlice,
1071                                         #self.users.DeletePersonFromSlice)
1072         #elif record.type == "authority":
1073             ## xxx TODO
1074             #pass
1075
1076 ### thierry
1077 # I don't think you plan on running a component manager at this point
1078 # let me clean up the mess of ComponentAPI that is deprecated anyways