4 from datetime import datetime
6 from sfa.util.faults import SliverDoesNotExist, UnknownSfaType
7 from sfa.util.sfalogging import logger
8 from sfa.storage.alchemy import dbsession
9 from sfa.storage.model import RegRecord, RegUser, RegSlice
10 from sqlalchemy.orm import joinedload
11 from sfa.trust.credential import Credential
14 from sfa.managers.driver import Driver
15 from sfa.rspecs.version_manager import VersionManager
16 from sfa.rspecs.rspec import RSpec
18 from sfa.util.xrn import Xrn, hrn_to_urn, get_authority
21 ## thierry: everything that is API-related (i.e. handling incoming requests)
23 # SlabDriver should be really only about talking to the senslab testbed
26 from sfa.senslab.OARrestapi import OARrestapi
27 from sfa.senslab.LDAPapi import LDAPapi
29 from sfa.senslab.slabpostgres import SlabDB, slab_dbsession, SenslabXP
32 from sfa.senslab.slabaggregate import SlabAggregate, slab_xrn_to_hostname, \
34 from sfa.senslab.slabslices import SlabSlices
39 # this inheritance scheme is so that the driver object can receive
40 # GetNodes or GetSites sorts of calls directly
41 # and thus minimize the differences in the managers with the pl version
45 class SlabDriver(Driver):
46 """ Senslab Driver class inherited from Driver generic class.
48 Contains methods compliant with the SFA standard and the testbed
49 infrastructure (calls to LDAP and OAR).
51 def __init__(self, config):
52 Driver.__init__ (self, config)
54 self.hrn = config.SFA_INTERFACE_HRN
55 self.root_auth = config.SFA_REGISTRY_ROOT_AUTH
56 self.oar = OARrestapi()
58 self.time_format = "%Y-%m-%d %H:%M:%S"
59 self.db = SlabDB(config, debug = False)
63 def sliver_status(self, slice_urn, slice_hrn):
64 """Receive a status request for slice named urn/hrn
65 urn:publicid:IDN+senslab+nturro_slice hrn senslab.nturro_slice
66 shall return a structure as described in
67 http://groups.geni.net/geni/wiki/GAPI_AM_API_V2#SliverStatus
68 NT : not sure if we should implement this or not, but used by sface.
72 #First get the slice with the slice hrn
73 slice_list = self.GetSlices(slice_filter = slice_hrn, \
74 slice_filter_type = 'slice_hrn')
76 if len(slice_list) is 0:
77 raise SliverDoesNotExist("%s slice_hrn" % (slice_hrn))
79 #Slice has the same slice hrn for each slice in the slice/lease list
80 #So fetch the info on the user once
81 one_slice = slice_list[0]
82 #recuser = dbsession.query(RegRecord).filter_by(record_id = \
83 #one_slice['record_id_user']).first()
85 #Make a list of all the nodes hostnames in use for this slice
87 for single_slice in slice_list:
88 for node in single_slice['node_ids']:
89 slice_nodes_list.append(node['hostname'])
91 #Get all the corresponding nodes details
92 nodes_all = self.GetNodes({'hostname':slice_nodes_list},
93 ['node_id', 'hostname','site','boot_state'])
94 nodeall_byhostname = dict([(one_node['hostname'], one_node) \
95 for one_node in nodes_all])
99 for single_slice in slice_list:
102 top_level_status = 'empty'
105 ['geni_urn','pl_login','geni_status','geni_resources'], None)
106 result['pl_login'] = one_slice['reg_researchers']['hrn']
107 logger.debug("Slabdriver - sliver_status Sliver status \
108 urn %s hrn %s single_slice %s \r\n " \
109 %(slice_urn, slice_hrn, single_slice))
111 nodes_in_slice = single_slice['node_ids']
114 result['geni_status'] = top_level_status
115 result['geni_resources'] = []
118 top_level_status = 'ready'
120 #A job is running on Senslab for this slice
121 # report about the local nodes that are in the slice only
123 result['geni_urn'] = slice_urn
127 #timestamp = float(sl['startTime']) + float(sl['walltime'])
128 #result['pl_expires'] = strftime(self.time_format, \
129 #gmtime(float(timestamp)))
130 #result['slab_expires'] = strftime(self.time_format,\
131 #gmtime(float(timestamp)))
134 for node in single_slice['node_ids']:
136 #res['slab_hostname'] = node['hostname']
137 #res['slab_boot_state'] = node['boot_state']
139 res['pl_hostname'] = node['hostname']
140 res['pl_boot_state'] = \
141 nodeall_byhostname[node['hostname']]['boot_state']
142 #res['pl_last_contact'] = strftime(self.time_format, \
143 #gmtime(float(timestamp)))
144 sliver_id = Xrn(slice_urn, type='slice', \
145 id=nodeall_byhostname[node['hostname']]['node_id'], \
146 authority=self.hrn).urn
148 res['geni_urn'] = sliver_id
149 node_name = node['hostname']
150 if nodeall_byhostname[node_name]['boot_state'] == 'Alive':
152 res['geni_status'] = 'ready'
154 res['geni_status'] = 'failed'
155 top_level_status = 'failed'
157 res['geni_error'] = ''
159 resources.append(res)
161 result['geni_status'] = top_level_status
162 result['geni_resources'] = resources
163 logger.debug("SLABDRIVER \tsliver_statusresources %s res %s "\
167 def get_user(self, hrn):
168 return dbsession.query(RegRecord).filter_by(hrn = hrn).first()
171 def create_sliver (self, slice_urn, slice_hrn, creds, rspec_string, \
173 aggregate = SlabAggregate(self)
175 slices = SlabSlices(self)
176 peer = slices.get_peer(slice_hrn)
177 sfa_peer = slices.get_sfa_peer(slice_hrn)
180 if not isinstance(creds, list):
184 slice_record = users[0].get('slice_record', {})
185 logger.debug("SLABDRIVER.PY \t ===============create_sliver \t\
186 creds %s \r\n \r\n users %s" \
188 slice_record['user'] = {'keys':users[0]['keys'], \
189 'email':users[0]['email'], \
190 'hrn':slice_record['reg-researchers'][0]}
192 rspec = RSpec(rspec_string)
193 logger.debug("SLABDRIVER.PY \t create_sliver \trspec.version \
194 %s slice_record %s users %s" \
195 %(rspec.version,slice_record, users))
198 # ensure site record exists?
199 # ensure slice record exists
200 #Removed options to verify_slice SA 14/08/12
201 sfa_slice = slices.verify_slice(slice_hrn, slice_record, peer, \
204 # ensure person records exists
205 #verify_persons returns added persons but since the return value
207 slices.verify_persons(slice_hrn, sfa_slice, users, peer, \
208 sfa_peer, options=options)
209 #requested_attributes returned by rspec.version.get_slice_attributes()
210 #unused, removed SA 13/08/12
211 rspec.version.get_slice_attributes()
213 logger.debug("SLABDRIVER.PY create_sliver slice %s " %(sfa_slice))
215 # add/remove slice from nodes
217 requested_slivers = [node.get('component_id') \
218 for node in rspec.version.get_nodes_with_slivers()\
219 if node.get('authority_id') is self.root_auth]
220 l = [ node for node in rspec.version.get_nodes_with_slivers() ]
221 logger.debug("SLADRIVER \tcreate_sliver requested_slivers \
222 requested_slivers %s listnodes %s" \
223 %(requested_slivers,l))
224 #verify_slice_nodes returns nodes, but unused here. Removed SA 13/08/12.
225 #slices.verify_slice_nodes(sfa_slice, requested_slivers, peer)
228 requested_lease_list = []
230 logger.debug("SLABDRIVER.PY \tcreate_sliver AVANTLEASE " )
231 rspec_requested_leases = rspec.version.get_leases()
232 for lease in rspec.version.get_leases():
233 single_requested_lease = {}
234 logger.debug("SLABDRIVER.PY \tcreate_sliver lease %s " %(lease))
236 if not lease.get('lease_id'):
237 if get_authority(lease['component_id']) == self.root_auth:
238 single_requested_lease['hostname'] = \
239 slab_xrn_to_hostname(\
240 lease.get('component_id').strip())
241 single_requested_lease['start_time'] = \
242 lease.get('start_time')
243 single_requested_lease['duration'] = lease.get('duration')
245 requested_lease_list.append(single_requested_lease)
247 logger.debug("SLABDRIVER.PY \tcreate_sliver APRESLEASE" )
248 #dCreate dict of leases by start_time, regrouping nodes reserved
250 #time, for the same amount of time = one job on OAR
251 requested_job_dict = {}
252 for lease in requested_lease_list:
254 #In case it is an asap experiment start_time is empty
255 if lease['start_time'] == '':
256 lease['start_time'] = '0'
258 if lease['start_time'] not in requested_job_dict:
259 if isinstance(lease['hostname'], str):
260 lease['hostname'] = [lease['hostname']]
262 requested_job_dict[lease['start_time']] = lease
265 job_lease = requested_job_dict[lease['start_time']]
266 if lease['duration'] == job_lease['duration'] :
267 job_lease['hostname'].append(lease['hostname'])
272 logger.debug("SLABDRIVER.PY \tcreate_sliver requested_job_dict %s "\
273 %(requested_job_dict))
274 #verify_slice_leases returns the leases , but the return value is unused
275 #here. Removed SA 13/08/12
276 slices.verify_slice_leases(sfa_slice, \
277 requested_job_dict, peer)
279 return aggregate.get_rspec(slice_xrn=slice_urn, version=rspec.version)
282 def delete_sliver (self, slice_urn, slice_hrn, creds, options):
284 sfa_slice_list = self.GetSlices(slice_filter = slice_hrn, \
285 slice_filter_type = 'slice_hrn')
287 if not sfa_slice_list:
290 #Delete all in the slice
291 for sfa_slice in sfa_slice_list:
294 logger.debug("SLABDRIVER.PY delete_sliver slice %s" %(sfa_slice))
295 slices = SlabSlices(self)
296 # determine if this is a peer slice
298 peer = slices.get_peer(slice_hrn)
299 #TODO delete_sliver SA : UnBindObjectFromPeer should be
300 #used when there is another
301 #senslab testbed, which is not the case 14/08/12 .
303 logger.debug("SLABDRIVER.PY delete_sliver peer %s" %(peer))
306 self.UnBindObjectFromPeer('slice', \
307 sfa_slice['record_id_slice'], \
309 self.DeleteSliceFromNodes(sfa_slice)
312 self.BindObjectToPeer('slice', \
313 sfa_slice['record_id_slice'], \
314 peer, sfa_slice['peer_slice_id'])
318 def AddSlice(self, slice_record, user_record):
319 """Add slice to the sfa tables and senslab table only if the user
320 already exists in senslab database(user already registered in LDAP).
321 There is no way to separate adding the slice to the tesbed
322 and then importing it from the testbed to SFA because of
323 senslab's architecture. Therefore, sfa tables are updated here.
326 sfa_record = RegSlice(hrn=slice_record['slice_hrn'],
327 gid=slice_record['gid'],
328 pointer=slice_record['slice_id'],
329 authority=slice_record['authority'])
331 logger.debug("SLABDRIVER.PY AddSlice sfa_record %s user_record %s" \
332 %(sfa_record, user_record))
333 sfa_record.just_created()
334 dbsession.add(sfa_record)
336 #Update the reg-researcher dependance table
337 sfa_record.reg_researchers = [user_record]
340 #Update the senslab table with the new slice
341 #slab_slice = SenslabXP( slice_hrn = slice_record['slice_hrn'], \
342 #record_id_slice = sfa_record.record_id , \
343 #record_id_user = slice_record['record_id_user'], \
344 #peer_authority = slice_record['peer_authority'])
346 #logger.debug("SLABDRIVER.PY \tAddSlice slice_record %s \
347 #slab_slice %s sfa_record %s" \
348 #%(slice_record,slab_slice, sfa_record))
349 #slab_dbsession.add(slab_slice)
350 #slab_dbsession.commit()
353 # first 2 args are None in case of resource discovery
354 def list_resources (self, slice_urn, slice_hrn, creds, options):
355 #cached_requested = options.get('cached', True)
357 version_manager = VersionManager()
358 # get the rspec's return format from options
360 version_manager.get_version(options.get('geni_rspec_version'))
361 version_string = "rspec_%s" % (rspec_version)
363 #panos adding the info option to the caching key (can be improved)
364 if options.get('info'):
365 version_string = version_string + "_" + \
366 options.get('info', 'default')
368 # Adding the list_leases option to the caching key
369 if options.get('list_leases'):
370 version_string = version_string + "_"+options.get('list_leases', 'default')
372 # Adding geni_available to caching key
373 if options.get('geni_available'):
374 version_string = version_string + "_" + str(options.get('geni_available'))
376 # look in cache first
377 #if cached_requested and self.cache and not slice_hrn:
378 #rspec = self.cache.get(version_string)
380 #logger.debug("SlabDriver.ListResources: \
381 #returning cached advertisement")
384 #panos: passing user-defined options
385 aggregate = SlabAggregate(self)
386 #origin_hrn = Credential(string=creds[0]).get_gid_caller().get_hrn()
387 #options.update({'origin_hrn':origin_hrn})
388 rspec = aggregate.get_rspec(slice_xrn=slice_urn, \
389 version=rspec_version, options=options)
392 #if self.cache and not slice_hrn:
393 #logger.debug("Slab.ListResources: stores advertisement in cache")
394 #self.cache.add(version_string, rspec)
399 def list_slices (self, creds, options):
400 # look in cache first
402 #slices = self.cache.get('slices')
404 #logger.debug("PlDriver.list_slices returns from cache")
409 slices = self.GetSlices()
410 logger.debug("SLABDRIVER.PY \tlist_slices hrn %s \r\n \r\n" %(slices))
411 slice_hrns = [slab_slice['hrn'] for slab_slice in slices]
413 slice_urns = [hrn_to_urn(slice_hrn, 'slice') \
414 for slice_hrn in slice_hrns]
418 #logger.debug ("SlabDriver.list_slices stores value in cache")
419 #self.cache.add('slices', slice_urns)
424 def register (self, sfa_record, hrn, pub_key):
426 Adding new user, slice, node or site should not be handled
430 Adding users = LDAP Senslab
431 Adding slice = Import from LDAP users
437 def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
438 """No site or node record update allowed in Senslab."""
440 pointer = old_sfa_record['pointer']
441 old_sfa_record_type = old_sfa_record['type']
443 # new_key implemented for users only
444 if new_key and old_sfa_record_type not in [ 'user' ]:
445 raise UnknownSfaType(old_sfa_record_type)
447 #if (type == "authority"):
448 #self.shell.UpdateSite(pointer, new_sfa_record)
450 if old_sfa_record_type == "slice":
451 slab_record = self.sfa_fields_to_slab_fields(old_sfa_record_type, \
453 if 'name' in slab_record:
454 slab_record.pop('name')
455 #Prototype should be UpdateSlice(self,
456 #auth, slice_id_or_name, slice_fields)
457 #Senslab cannot update slice since slice = job
458 #so we must delete and create another job
459 self.UpdateSlice(pointer, slab_record)
461 elif old_sfa_record_type == "user":
463 all_fields = new_sfa_record
464 for key in all_fields.keys():
465 if key in ['first_name', 'last_name', 'title', 'email',
466 'password', 'phone', 'url', 'bio', 'accepted_aup',
468 update_fields[key] = all_fields[key]
469 self.UpdatePerson(pointer, update_fields)
472 # must check this key against the previous one if it exists
473 persons = self.GetPersons([pointer], ['key_ids'])
475 keys = person['key_ids']
476 keys = self.GetKeys(person['key_ids'])
478 # Delete all stale keys
481 if new_key != key['key']:
482 self.DeleteKey(key['key_id'])
486 self.AddPersonKey(pointer, {'key_type': 'ssh', \
493 def remove (self, sfa_record):
494 sfa_record_type = sfa_record['type']
495 hrn = sfa_record['hrn']
496 if sfa_record_type == 'user':
498 #get user from senslab ldap
499 person = self.GetPersons(sfa_record)
500 #No registering at a given site in Senslab.
501 #Once registered to the LDAP, all senslab sites are
504 #Mark account as disabled in ldap
505 self.DeletePerson(sfa_record)
506 elif sfa_record_type == 'slice':
507 if self.GetSlices(slice_filter = hrn, \
508 slice_filter_type = 'slice_hrn'):
509 self.DeleteSlice(sfa_record)
511 #elif type == 'authority':
512 #if self.GetSites(pointer):
513 #self.DeleteSite(pointer)
519 #TODO clean GetPeers. 05/07/12SA
520 def GetPeers (self, auth = None, peer_filter=None, return_fields_list=None):
522 existing_records = {}
523 existing_hrns_by_types = {}
524 logger.debug("SLABDRIVER \tGetPeers auth = %s, peer_filter %s, \
525 return_field %s " %(auth , peer_filter, return_fields_list))
526 all_records = dbsession.query(RegRecord).filter(RegRecord.type.like('%authority%')).all()
528 for record in all_records:
529 existing_records[(record.hrn, record.type)] = record
530 if record.type not in existing_hrns_by_types:
531 existing_hrns_by_types[record.type] = [record.hrn]
533 existing_hrns_by_types[record.type].append(record.hrn)
536 logger.debug("SLABDRIVER \tGetPeer\texisting_hrns_by_types %s "\
537 %( existing_hrns_by_types))
542 records_list.append(existing_records[(peer_filter,'authority')])
544 for hrn in existing_hrns_by_types['authority']:
545 records_list.append(existing_records[(hrn,'authority')])
547 logger.debug("SLABDRIVER \tGetPeer \trecords_list %s " \
553 return_records = records_list
554 if not peer_filter and not return_fields_list:
558 logger.debug("SLABDRIVER \tGetPeer return_records %s " \
560 return return_records
563 #TODO : Handling OR request in make_ldap_filters_from_records
564 #instead of the for loop
565 #over the records' list
566 def GetPersons(self, person_filter=None):
568 person_filter should be a list of dictionnaries when not set to None.
569 Returns a list of users whose accounts are enabled found in ldap.
572 logger.debug("SLABDRIVER \tGetPersons person_filter %s" \
575 if person_filter and isinstance(person_filter, list):
576 #If we are looking for a list of users (list of dict records)
577 #Usually the list contains only one user record
578 for searched_attributes in person_filter:
580 #Get only enabled user accounts in senslab LDAP :
581 #add a filter for make_ldap_filters_from_record
582 person = self.ldap.LdapFindUser(searched_attributes, \
583 is_user_enabled=True)
584 #If a person was found, append it to the list
586 person_list.append(person)
588 #If the list is empty, return None
589 if len(person_list) is 0:
593 #Get only enabled user accounts in senslab LDAP :
594 #add a filter for make_ldap_filters_from_record
595 person_list = self.ldap.LdapFindUser(is_user_enabled=True)
599 def GetTimezone(self):
600 """ Get the OAR servier time and timezone.
601 Unused SA 16/11/12"""
602 server_timestamp, server_tz = self.oar.parser.\
603 SendRequest("GET_timezone")
604 return server_timestamp, server_tz
607 def DeleteJobs(self, job_id, slice_hrn):
608 if not job_id or job_id is -1:
610 username = slice_hrn.split(".")[-1].rstrip("_slice")
612 reqdict['method'] = "delete"
613 reqdict['strval'] = str(job_id)
616 answer = self.oar.POSTRequestToOARRestAPI('DELETE_jobs_id', \
618 logger.debug("SLABDRIVER \tDeleteJobs jobid %s \r\n answer %s \
619 username %s" %(job_id,answer, username))
624 ##TODO : Unused GetJobsId ? SA 05/07/12
625 #def GetJobsId(self, job_id, username = None ):
627 #Details about a specific job.
628 #Includes details about submission time, jot type, state, events,
629 #owner, assigned ressources, walltime etc...
633 #node_list_k = 'assigned_network_address'
634 ##Get job info from OAR
635 #job_info = self.oar.parser.SendRequest(req, job_id, username)
637 #logger.debug("SLABDRIVER \t GetJobsId %s " %(job_info))
639 #if job_info['state'] == 'Terminated':
640 #logger.debug("SLABDRIVER \t GetJobsId job %s TERMINATED"\
643 #if job_info['state'] == 'Error':
644 #logger.debug("SLABDRIVER \t GetJobsId ERROR message %s "\
649 #logger.error("SLABDRIVER \tGetJobsId KeyError")
652 #parsed_job_info = self.get_info_on_reserved_nodes(job_info, \
654 ##Replaces the previous entry
655 ##"assigned_network_address" / "reserved_resources"
657 #job_info.update({'node_ids':parsed_job_info[node_list_k]})
658 #del job_info[node_list_k]
659 #logger.debug(" \r\nSLABDRIVER \t GetJobsId job_info %s " %(job_info))
663 def GetJobsResources(self, job_id, username = None):
664 #job_resources=['reserved_resources', 'assigned_resources',\
665 #'job_id', 'job_uri', 'assigned_nodes',\
667 #assigned_res = ['resource_id', 'resource_uri']
668 #assigned_n = ['node', 'node_uri']
670 req = "GET_jobs_id_resources"
673 #Get job resources list from OAR
674 node_id_list = self.oar.parser.SendRequest(req, job_id, username)
675 logger.debug("SLABDRIVER \t GetJobsResources %s " %(node_id_list))
678 self.__get_hostnames_from_oar_node_ids(node_id_list)
681 #Replaces the previous entry "assigned_network_address" /
682 #"reserved_resources"
684 job_info = {'node_ids': hostname_list}
689 def get_info_on_reserved_nodes(self, job_info, node_list_name):
690 #Get the list of the testbed nodes records and make a
691 #dictionnary keyed on the hostname out of it
692 node_list_dict = self.GetNodes()
693 #node_hostname_list = []
694 node_hostname_list = [node['hostname'] for node in node_list_dict]
695 #for node in node_list_dict:
696 #node_hostname_list.append(node['hostname'])
697 node_dict = dict(zip(node_hostname_list, node_list_dict))
699 reserved_node_hostname_list = []
700 for index in range(len(job_info[node_list_name])):
701 #job_info[node_list_name][k] =
702 reserved_node_hostname_list[index] = \
703 node_dict[job_info[node_list_name][index]]['hostname']
705 logger.debug("SLABDRIVER \t get_info_on_reserved_nodes \
706 reserved_node_hostname_list %s" \
707 %(reserved_node_hostname_list))
709 logger.error("SLABDRIVER \t get_info_on_reserved_nodes KEYERROR " )
711 return reserved_node_hostname_list
713 def GetNodesCurrentlyInUse(self):
714 """Returns a list of all the nodes already involved in an oar job"""
715 return self.oar.parser.SendRequest("GET_running_jobs")
717 def __get_hostnames_from_oar_node_ids(self, resource_id_list ):
718 full_nodes_dict_list = self.GetNodes()
719 #Put the full node list into a dictionary keyed by oar node id
720 oar_id_node_dict = {}
721 for node in full_nodes_dict_list:
722 oar_id_node_dict[node['oar_id']] = node
724 #logger.debug("SLABDRIVER \t __get_hostnames_from_oar_node_ids\
725 #oar_id_node_dict %s" %(oar_id_node_dict))
727 hostname_dict_list = []
728 for resource_id in resource_id_list:
729 #Because jobs requested "asap" do not have defined resources
730 if resource_id is not "Undefined":
731 hostname_dict_list.append(\
732 oar_id_node_dict[resource_id]['hostname'])
734 #hostname_list.append(oar_id_node_dict[resource_id]['hostname'])
735 return hostname_dict_list
737 def GetReservedNodes(self,username = None):
738 #Get the nodes in use and the reserved nodes
739 reservation_dict_list = \
740 self.oar.parser.SendRequest("GET_reserved_nodes", \
744 for resa in reservation_dict_list:
745 logger.debug ("GetReservedNodes resa %s"%(resa))
746 #dict list of hostnames and their site
747 resa['reserved_nodes'] = \
748 self.__get_hostnames_from_oar_node_ids(resa['resource_ids'])
750 #del resa['resource_ids']
751 return reservation_dict_list
753 def GetNodes(self, node_filter_dict = None, return_fields_list = None):
755 node_filter_dict : dictionnary of lists
758 node_dict_by_id = self.oar.parser.SendRequest("GET_resources_full")
759 node_dict_list = node_dict_by_id.values()
760 logger.debug (" SLABDRIVER GetNodes node_filter_dict %s \
761 return_fields_list %s "%(node_filter_dict, return_fields_list))
762 #No filtering needed return the list directly
763 if not (node_filter_dict or return_fields_list):
764 return node_dict_list
766 return_node_list = []
768 for filter_key in node_filter_dict:
770 #Filter the node_dict_list by each value contained in the
771 #list node_filter_dict[filter_key]
772 for value in node_filter_dict[filter_key]:
773 for node in node_dict_list:
774 if node[filter_key] == value:
775 if return_fields_list :
777 for k in return_fields_list:
779 return_node_list.append(tmp)
781 return_node_list.append(node)
783 logger.log_exc("GetNodes KeyError")
787 return return_node_list
790 def GetSites(self, site_filter_name_list = None, return_fields_list = None):
791 site_dict = self.oar.parser.SendRequest("GET_sites")
792 #site_dict : dict where the key is the sit ename
793 return_site_list = []
794 if not ( site_filter_name_list or return_fields_list):
795 return_site_list = site_dict.values()
796 return return_site_list
798 for site_filter_name in site_filter_name_list:
799 if site_filter_name in site_dict:
800 if return_fields_list:
801 for field in return_fields_list:
804 tmp[field] = site_dict[site_filter_name][field]
806 logger.error("GetSites KeyError %s "%(field))
808 return_site_list.append(tmp)
810 return_site_list.append( site_dict[site_filter_name])
813 return return_site_list
816 def _sql_get_slice_info( self, slice_filter ):
817 #DO NOT USE RegSlice - reg_researchers to get the hrn
818 #of the user otherwise will mess up the RegRecord in
819 #Resolve, don't know why - SA 08/08/2012
821 #Only one entry for one user = one slice in slab_xp table
822 #slicerec = dbsession.query(RegRecord).filter_by(hrn = slice_filter).first()
823 raw_slicerec = dbsession.query(RegSlice).options(joinedload('reg_researchers')).filter_by(hrn = slice_filter).first()
824 #raw_slicerec = dbsession.query(RegRecord).filter_by(hrn = slice_filter).first()
827 #raw_slicerec.reg_researchers
828 raw_slicerec = raw_slicerec.__dict__
829 logger.debug(" SLABDRIVER \t get_slice_info slice_filter %s raw_slicerec %s"%(slice_filter,raw_slicerec))
830 slicerec = raw_slicerec
831 #only one researcher per slice so take the first one
832 #slicerec['reg_researchers'] = raw_slicerec['reg_researchers']
833 #del slicerec['reg_researchers']['_sa_instance_state']
840 def _sql_get_slice_info_from_user( self, slice_filter ):
841 #slicerec = dbsession.query(RegRecord).filter_by(record_id = slice_filter).first()
842 raw_slicerec = dbsession.query(RegUser).options(joinedload('reg_slices_as_researcher')).filter_by(record_id = slice_filter).first()
843 #raw_slicerec = dbsession.query(RegRecord).filter_by(record_id = slice_filter).first()
844 #Put it in correct order
845 user_needed_fields = ['peer_authority', 'hrn', 'last_updated', 'classtype', 'authority', 'gid', 'record_id', 'date_created', 'type', 'email', 'pointer']
846 slice_needed_fields = ['peer_authority', 'hrn', 'last_updated', 'classtype', 'authority', 'gid', 'record_id', 'date_created', 'type', 'pointer']
848 #raw_slicerec.reg_slices_as_researcher
849 raw_slicerec = raw_slicerec.__dict__
851 slicerec = dict([(k,raw_slicerec['reg_slices_as_researcher'][0].__dict__[k]) for k in slice_needed_fields])
852 slicerec['reg_researchers'] = dict([(k, raw_slicerec[k]) for k in user_needed_fields])
853 #TODO Handle multiple slices for one user SA 10/12/12
854 #for now only take the first slice record associated to the rec user
855 ##slicerec = raw_slicerec['reg_slices_as_researcher'][0].__dict__
856 #del raw_slicerec['reg_slices_as_researcher']
857 #slicerec['reg_researchers'] = raw_slicerec
858 ##del slicerec['_sa_instance_state']
865 def _get_slice_records(self, slice_filter = None, \
866 slice_filter_type = None):
869 #Get list of slices based on the slice hrn
870 if slice_filter_type == 'slice_hrn':
872 if get_authority(slice_filter) == self.root_auth:
873 login = slice_filter.split(".")[1].split("_")[0]
875 slicerec = self._sql_get_slice_info(slice_filter)
880 #Get slice based on user id
881 if slice_filter_type == 'record_id_user':
883 slicerec = self._sql_get_slice_info_from_user(slice_filter)
886 fixed_slicerec_dict = slicerec
887 #At this point if the there is no login it means
888 #record_id_user filter has been used for filtering
890 #If theslice record is from senslab
891 if fixed_slicerec_dict['peer_authority'] is None:
892 login = fixed_slicerec_dict['hrn'].split(".")[1].split("_")[0]
893 return login, fixed_slicerec_dict
896 def GetSlices(self, slice_filter = None, slice_filter_type = None):
897 """ Get the slice records from the slab db.
898 Returns a slice ditc if slice_filter and slice_filter_type
900 Returns a list of slice dictionnaries if there are no filters
905 authorized_filter_types_list = ['slice_hrn', 'record_id_user']
906 return_slicerec_dictlist = []
908 #First try to get information on the slice based on the filter provided
909 if slice_filter_type in authorized_filter_types_list:
911 login, fixed_slicerec_dict = \
912 self._get_slice_records(slice_filter, slice_filter_type)
913 logger.debug(" SLABDRIVER \tGetSlices login %s \
914 slice record %s slice_filter %s slice_filter_type %s "\
915 %(login, fixed_slicerec_dict,slice_filter, slice_filter_type))
918 #Now we have the slice record fixed_slicerec_dict, get the
919 #jobs associated to this slice
920 #leases_list = self.GetReservedNodes(username = login)
921 leases_list = self.GetLeases(login = login)
922 #If no job is running or no job scheduled
923 #return only the slice record
924 if leases_list == [] and fixed_slicerec_dict:
925 return_slicerec_dictlist.append(fixed_slicerec_dict)
927 #If several jobs for one slice , put the slice record into
928 # each lease information dict
929 for lease in leases_list :
932 reserved_list = lease['reserved_nodes']
934 slicerec_dict['oar_job_id']= lease['lease_id']
935 slicerec_dict.update({'list_node_ids':{'hostname':reserved_list}})
936 slicerec_dict.update({'node_ids':lease['reserved_nodes']})
938 #Update lease dict with the slice record
939 if fixed_slicerec_dict:
940 fixed_slicerec_dict['oar_job_id'] = []
941 fixed_slicerec_dict['oar_job_id'].append(slicerec_dict['oar_job_id'])
942 slicerec_dict.update(fixed_slicerec_dict)
943 #slicerec_dict.update({'hrn':\
944 #str(fixed_slicerec_dict['slice_hrn'])})
947 return_slicerec_dictlist.append(slicerec_dict)
948 logger.debug("SLABDRIVER.PY \tGetSlices \
949 slicerec_dict %s return_slicerec_dictlist %s \
950 lease['reserved_nodes'] \
951 %s" %(slicerec_dict, return_slicerec_dictlist, \
952 lease['reserved_nodes'] ))
954 logger.debug("SLABDRIVER.PY \tGetSlices RETURN \
955 return_slicerec_dictlist %s" \
956 %(return_slicerec_dictlist))
958 return return_slicerec_dictlist
962 #Get all slices from the senslab sfa database ,
963 #put them in dict format
964 #query_slice_list = dbsession.query(RegRecord).all()
965 query_slice_list = dbsession.query(RegSlice).options(joinedload('reg_researchers')).all()
966 #query_slice_list = dbsession.query(RegRecord).filter_by(type='slice').all()
967 #query_slice_list = slab_dbsession.query(SenslabXP).all()
968 return_slicerec_dictlist = []
969 for record in query_slice_list:
970 tmp = record.__dict__
971 tmp['reg_researchers'] = tmp['reg_researchers'][0].__dict__
972 #del tmp['reg_researchers']['_sa_instance_state']
973 return_slicerec_dictlist.append(tmp)
974 #return_slicerec_dictlist.append(record.__dict__)
976 #Get all the jobs reserved nodes
977 leases_list = self.GetReservedNodes()
980 for fixed_slicerec_dict in return_slicerec_dictlist:
982 #Check if the slice belongs to a senslab user
983 if fixed_slicerec_dict['peer_authority'] is None:
984 owner = fixed_slicerec_dict['hrn'].split(".")[1].split("_")[0]
987 for lease in leases_list:
988 if owner == lease['user']:
989 slicerec_dict['oar_job_id'] = lease['lease_id']
991 #for reserved_node in lease['reserved_nodes']:
992 logger.debug("SLABDRIVER.PY \tGetSlices lease %s "\
995 reserved_list = lease['reserved_nodes']
997 slicerec_dict.update({'node_ids':lease['reserved_nodes']})
998 slicerec_dict.update({'list_node_ids':{'hostname':reserved_list}})
999 slicerec_dict.update(fixed_slicerec_dict)
1000 #slicerec_dict.update({'hrn':\
1001 #str(fixed_slicerec_dict['slice_hrn'])})
1002 #return_slicerec_dictlist.append(slicerec_dict)
1003 fixed_slicerec_dict.update(slicerec_dict)
1005 logger.debug("SLABDRIVER.PY \tGetSlices RETURN \
1006 return_slicerec_dictlist %s \slice_filter %s " \
1007 %(return_slicerec_dictlist, slice_filter))
1009 return return_slicerec_dictlist
1012 def testbed_name (self): return self.hrn
1014 # 'geni_request_rspec_versions' and 'geni_ad_rspec_versions' are mandatory
1015 def aggregate_version (self):
1016 version_manager = VersionManager()
1017 ad_rspec_versions = []
1018 request_rspec_versions = []
1019 for rspec_version in version_manager.versions:
1020 if rspec_version.content_type in ['*', 'ad']:
1021 ad_rspec_versions.append(rspec_version.to_dict())
1022 if rspec_version.content_type in ['*', 'request']:
1023 request_rspec_versions.append(rspec_version.to_dict())
1025 'testbed':self.testbed_name(),
1026 'geni_request_rspec_versions': request_rspec_versions,
1027 'geni_ad_rspec_versions': ad_rspec_versions,
1033 # Convert SFA fields to PLC fields for use when registering up updating
1034 # registry record in the PLC database
1036 # @param type type of record (user, slice, ...)
1037 # @param hrn human readable name
1038 # @param sfa_fields dictionary of SFA fields
1039 # @param slab_fields dictionary of PLC fields (output)
1041 def sfa_fields_to_slab_fields(self, sfa_type, hrn, record):
1045 #for field in record:
1046 # slab_record[field] = record[field]
1048 if sfa_type == "slice":
1049 #instantion used in get_slivers ?
1050 if not "instantiation" in slab_record:
1051 slab_record["instantiation"] = "senslab-instantiated"
1052 #slab_record["hrn"] = hrn_to_pl_slicename(hrn)
1053 #Unused hrn_to_pl_slicename because Slab's hrn already
1054 #in the appropriate form SA 23/07/12
1055 slab_record["hrn"] = hrn
1056 logger.debug("SLABDRIVER.PY sfa_fields_to_slab_fields \
1057 slab_record %s " %(slab_record['hrn']))
1059 slab_record["url"] = record["url"]
1060 if "description" in record:
1061 slab_record["description"] = record["description"]
1062 if "expires" in record:
1063 slab_record["expires"] = int(record["expires"])
1065 #nodes added by OAR only and then imported to SFA
1066 #elif type == "node":
1067 #if not "hostname" in slab_record:
1068 #if not "hostname" in record:
1069 #raise MissingSfaInfo("hostname")
1070 #slab_record["hostname"] = record["hostname"]
1071 #if not "model" in slab_record:
1072 #slab_record["model"] = "geni"
1075 #elif type == "authority":
1076 #slab_record["login_base"] = hrn_to_slab_login_base(hrn)
1078 #if not "name" in slab_record:
1079 #slab_record["name"] = hrn
1081 #if not "abbreviated_name" in slab_record:
1082 #slab_record["abbreviated_name"] = hrn
1084 #if not "enabled" in slab_record:
1085 #slab_record["enabled"] = True
1087 #if not "is_public" in slab_record:
1088 #slab_record["is_public"] = True
1095 def __transforms_timestamp_into_date(self, xp_utc_timestamp = None):
1096 """ Transforms unix timestamp into valid OAR date format """
1098 #Used in case of a scheduled experiment (not immediate)
1099 #To run an XP immediately, don't specify date and time in RSpec
1100 #They will be set to None.
1101 if xp_utc_timestamp:
1102 #transform the xp_utc_timestamp into server readable time
1103 xp_server_readable_date = datetime.fromtimestamp(int(\
1104 xp_utc_timestamp)).strftime(self.time_format)
1106 return xp_server_readable_date
1114 def LaunchExperimentOnOAR(self, added_nodes, slice_name, \
1115 lease_start_time, lease_duration, slice_user=None):
1117 lease_dict['lease_start_time'] = lease_start_time
1118 lease_dict['lease_duration'] = lease_duration
1119 lease_dict['added_nodes'] = added_nodes
1120 lease_dict['slice_name'] = slice_name
1121 lease_dict['slice_user'] = slice_user
1122 lease_dict['grain'] = self.GetLeaseGranularity()
1123 lease_dict['time_format'] = self.time_format
1126 def __create_job_structure_request_for_OAR(lease_dict):
1127 """ Creates the structure needed for a correct POST on OAR.
1128 Makes the timestamp transformation into the appropriate format.
1129 Sends the POST request to create the job with the resources in
1138 reqdict['workdir'] = '/tmp'
1139 reqdict['resource'] = "{network_address in ("
1141 for node in lease_dict['added_nodes']:
1142 logger.debug("\r\n \r\n OARrestapi \t \
1143 __create_job_structure_request_for_OAR node %s" %(node))
1145 # Get the ID of the node
1147 reqdict['resource'] += "'" + nodeid + "', "
1148 nodeid_list.append(nodeid)
1150 custom_length = len(reqdict['resource'])- 2
1151 reqdict['resource'] = reqdict['resource'][0:custom_length] + \
1152 ")}/nodes=" + str(len(nodeid_list))
1154 def __process_walltime(duration):
1155 """ Calculates the walltime in seconds from the duration in H:M:S
1156 specified in the RSpec.
1160 # Fixing the walltime by adding a few delays.
1161 # First put the walltime in seconds oarAdditionalDelay = 20;
1162 # additional delay for /bin/sleep command to
1163 # take in account prologue and epilogue scripts execution
1164 # int walltimeAdditionalDelay = 240; additional delay
1165 desired_walltime = duration
1166 total_walltime = desired_walltime + 240 #+4 min Update SA 23/10/12
1167 sleep_walltime = desired_walltime # 0 sec added Update SA 23/10/12
1169 #Put the walltime back in str form
1170 #First get the hours
1171 walltime.append(str(total_walltime / 3600))
1172 total_walltime = total_walltime - 3600 * int(walltime[0])
1173 #Get the remaining minutes
1174 walltime.append(str(total_walltime / 60))
1175 total_walltime = total_walltime - 60 * int(walltime[1])
1177 walltime.append(str(total_walltime))
1180 logger.log_exc(" __process_walltime duration null")
1182 return walltime, sleep_walltime
1185 walltime, sleep_walltime = \
1186 __process_walltime(int(lease_dict['lease_duration'])*lease_dict['grain'])
1189 reqdict['resource'] += ",walltime=" + str(walltime[0]) + \
1190 ":" + str(walltime[1]) + ":" + str(walltime[2])
1191 reqdict['script_path'] = "/bin/sleep " + str(sleep_walltime)
1193 #In case of a scheduled experiment (not immediate)
1194 #To run an XP immediately, don't specify date and time in RSpec
1195 #They will be set to None.
1196 if lease_dict['lease_start_time'] is not '0':
1197 #Readable time accepted by OAR
1198 start_time = datetime.fromtimestamp(int(lease_dict['lease_start_time'])).\
1199 strftime(lease_dict['time_format'])
1200 reqdict['reservation'] = start_time
1201 #If there is not start time, Immediate XP. No need to add special
1205 reqdict['type'] = "deploy"
1206 reqdict['directory'] = ""
1207 reqdict['name'] = "SFA_" + lease_dict['slice_user']
1212 #Create the request for OAR
1213 reqdict = __create_job_structure_request_for_OAR(lease_dict)
1214 # first step : start the OAR job and update the job
1215 logger.debug("SLABDRIVER.PY \tLaunchExperimentOnOAR reqdict %s\
1218 answer = self.oar.POSTRequestToOARRestAPI('POST_job', \
1219 reqdict, slice_user)
1220 logger.debug("SLABDRIVER \tLaunchExperimentOnOAR jobid %s " %(answer))
1222 jobid = answer['id']
1224 logger.log_exc("SLABDRIVER \tLaunchExperimentOnOAR \
1225 Impossible to create job %s " %(answer))
1229 def __configure_experiment(jobid, added_nodes):
1230 # second step : configure the experiment
1231 # we need to store the nodes in a yaml (well...) file like this :
1232 # [1,56,23,14,45,75] with name /tmp/sfa<jobid>.json
1233 tmp_dir = '/tmp/sfa/'
1234 if not os.path.exists(tmp_dir):
1235 os.makedirs(tmp_dir)
1236 job_file = open(tmp_dir + str(jobid) + '.json', 'w')
1238 job_file.write(str(added_nodes[0].strip('node')))
1239 for node in added_nodes[1:len(added_nodes)] :
1240 job_file.write(', '+ node.strip('node'))
1245 def __launch_senslab_experiment(jobid):
1246 # third step : call the senslab-experiment wrapper
1247 #command= "java -jar target/sfa-1.0-jar-with-dependencies.jar
1248 # "+str(jobid)+" "+slice_user
1249 javacmdline = "/usr/bin/java"
1251 "/opt/senslabexperimentwrapper/sfa-1.0-jar-with-dependencies.jar"
1253 output = subprocess.Popen([javacmdline, "-jar", jarname, str(jobid), \
1254 slice_user],stdout=subprocess.PIPE).communicate()[0]
1256 logger.debug("SLABDRIVER \t __configure_experiment wrapper returns%s " \
1263 logger.debug("SLABDRIVER \tLaunchExperimentOnOAR jobid %s \
1264 added_nodes %s slice_user %s" %(jobid, added_nodes, slice_user))
1267 __configure_experiment(jobid, added_nodes)
1268 __launch_senslab_experiment(jobid)
1273 def AddLeases(self, hostname_list, slice_record, \
1274 lease_start_time, lease_duration):
1275 logger.debug("SLABDRIVER \r\n \r\n \t AddLeases hostname_list %s \
1276 slice_record %s lease_start_time %s lease_duration %s "\
1277 %( hostname_list, slice_record , lease_start_time, \
1280 #tmp = slice_record['reg-researchers'][0].split(".")
1281 username = slice_record['user']['uid']
1282 #username = tmp[(len(tmp)-1)]
1283 job_id = self.LaunchExperimentOnOAR(hostname_list, slice_record['hrn'], \
1284 lease_start_time, lease_duration, username)
1285 start_time = datetime.fromtimestamp(int(lease_start_time)).strftime(self.time_format)
1286 end_time = lease_start_time + lease_duration
1287 logger.debug("SLABDRIVER \r\n \r\n \t AddLeases %s %s %s " %(type(slice_record['hrn']), type(job_id), type(end_time)))
1288 slab_ex_row = SenslabXP(slice_record['hrn'], job_id, end_time)
1289 logger.debug("SLABDRIVER \r\n \r\n \t slab_ex_row %s" %(slab_ex_row))
1290 slab_dbsession.add(slab_ex_row)
1291 slab_dbsession.commit()
1293 logger.debug("SLABDRIVER \t AddLeases hostname_list start_time %s " %(start_time))
1298 #Delete the jobs from job_senslab table
1299 def DeleteSliceFromNodes(self, slice_record):
1300 for job_id in slice_record['oar_job_id']:
1301 self.DeleteJobs(job_id, slice_record['hrn'])
1305 def GetLeaseGranularity(self):
1306 """ Returns the granularity of Senslab testbed.
1307 OAR returns seconds for experiments duration.
1308 Defined in seconds. """
1313 def update_jobs_in_slabdb(self, job_oar_list, jobs_psql):
1314 #Get all the entries in slab_xp table
1317 jobs_psql = set(jobs_psql)
1318 kept_jobs = set(job_oar_list).intersection(jobs_psql)
1320 deleted_jobs = set(jobs_psql).difference(kept_jobs)
1321 deleted_jobs = list(deleted_jobs)
1322 if len(deleted_jobs) > 0:
1323 slab_dbsession.query(SenslabXP).filter(SenslabXP.job_id.in_(deleted_jobs)).delete(synchronize_session='fetch')
1324 slab_dbsession.commit()
1330 def GetLeases(self, lease_filter_dict=None, login=None):
1333 unfiltered_reservation_list = self.GetReservedNodes(login)
1335 reservation_list = []
1336 #Find the slice associated with this user senslab ldap uid
1337 logger.debug(" SLABDRIVER.PY \tGetLeases unfiltered_reservation_list %s " %(unfiltered_reservation_list))
1338 #Create user dict first to avoid looking several times for
1339 #the same user in LDAP SA 27/07/12
1343 jobs_psql_query = slab_dbsession.query(SenslabXP).all()
1344 jobs_psql_dict = [ (row.job_id, row.__dict__ )for row in jobs_psql_query ]
1345 jobs_psql_dict = dict(jobs_psql_dict)
1346 logger.debug("SLABDRIVER \tGetLeases jobs_psql_dict %s"\
1348 jobs_psql_id_list = [ row.job_id for row in jobs_psql_query ]
1352 for resa in unfiltered_reservation_list:
1353 logger.debug("SLABDRIVER \tGetLeases USER %s"\
1355 #Cosntruct list of jobs (runing, waiting..) in oar
1356 job_oar_list.append(resa['lease_id'])
1357 #If there is information on the job in SLAB DB (slice used and job id)
1358 if resa['lease_id'] in jobs_psql_dict:
1359 job_info = jobs_psql_dict[resa['lease_id']]
1360 logger.debug("SLABDRIVER \tGetLeases resa_user_dict %s"\
1362 resa['slice_hrn'] = job_info['slice_hrn']
1363 resa['slice_id'] = hrn_to_urn(resa['slice_hrn'], 'slice')
1365 #Assume it is a senslab slice:
1367 resa['slice_id'] = hrn_to_urn(self.root_auth+'.'+ resa['user'] +"_slice" , 'slice')
1368 #if resa['user'] not in resa_user_dict:
1369 #logger.debug("SLABDRIVER \tGetLeases userNOTIN ")
1370 #ldap_info = self.ldap.LdapSearch('(uid='+resa['user']+')')
1372 #ldap_info = ldap_info[0][1]
1373 ##Get the backref :relationship table reg-researchers
1374 #user = dbsession.query(RegUser).options(joinedload('reg_slices_as_researcher')).filter_by(email = \
1375 #ldap_info['mail'][0])
1377 #user = user.first()
1378 #user = user.__dict__
1379 #slice_info = user['reg_slices_as_researcher'][0].__dict__
1380 ##Separated in case user not in database :
1381 ##record_id not defined SA 17/07//12
1383 ##query_slice_info = slab_dbsession.query(SenslabXP).filter_by(record_id_user = user.record_id)
1384 ##if query_slice_info:
1385 ##slice_info = query_slice_info.first()
1389 #resa_user_dict[resa['user']] = {}
1390 #resa_user_dict[resa['user']]['ldap_info'] = user
1391 #resa_user_dict[resa['user']]['slice_info'] = slice_info
1393 #resa['slice_hrn'] = resa_user_dict[resa['user']]['slice_info']['hrn']
1394 #resa['slice_id'] = hrn_to_urn(resa['slice_hrn'], 'slice')
1397 resa['component_id_list'] = []
1398 resa['hrn'] = Xrn(resa['slice_id']).get_hrn()
1399 #Transform the hostnames into urns (component ids)
1400 for node in resa['reserved_nodes']:
1401 #resa['component_id_list'].append(hostname_to_urn(self.hrn, \
1402 #self.root_auth, node['hostname']))
1403 slab_xrn = slab_xrn_object(self.root_auth, node)
1404 resa['component_id_list'].append(slab_xrn.urn)
1406 if lease_filter_dict:
1407 logger.debug("SLABDRIVER \tGetLeases resa_ %s \r\n leasefilter %s"\
1408 %(resa,lease_filter_dict))
1410 if lease_filter_dict['name'] == resa['slice_hrn']:
1411 reservation_list.append(resa)
1413 if lease_filter_dict is None:
1414 reservation_list = unfiltered_reservation_list
1416 #del unfiltered_reservation_list[unfiltered_reservation_list.index(resa)]
1419 self.update_jobs_in_slabdb(job_oar_list, jobs_psql_id_list)
1421 #for resa in unfiltered_reservation_list:
1425 #if resa['user'] in resa_user_dict:
1426 #resa['slice_hrn'] = resa_user_dict[resa['user']]['slice_info']['hrn']
1427 #resa['slice_id'] = hrn_to_urn(resa['slice_hrn'], 'slice')
1429 ##resa['slice_id'] = hrn_to_urn(slice_info.slice_hrn, 'slice')
1430 #resa['component_id_list'] = []
1431 ##Transform the hostnames into urns (component ids)
1432 #for node in resa['reserved_nodes']:
1433 ##resa['component_id_list'].append(hostname_to_urn(self.hrn, \
1434 ##self.root_auth, node['hostname']))
1435 #slab_xrn = slab_xrn_object(self.root_auth, node)
1436 #resa['component_id_list'].append(slab_xrn.urn)
1438 ##Filter the reservation list if necessary
1439 ##Returns all the leases associated with a given slice
1440 #if lease_filter_dict:
1441 #logger.debug("SLABDRIVER \tGetLeases lease_filter_dict %s"\
1442 #%(lease_filter_dict))
1443 #for resa in unfiltered_reservation_list:
1444 #if lease_filter_dict['name'] == resa['slice_hrn']:
1445 #reservation_list.append(resa)
1447 #reservation_list = unfiltered_reservation_list
1449 logger.debug(" SLABDRIVER.PY \tGetLeases reservation_list %s"\
1450 %(reservation_list))
1451 return reservation_list
1453 def augment_records_with_testbed_info (self, sfa_records):
1454 return self.fill_record_info (sfa_records)
1456 def fill_record_info(self, record_list):
1458 Given a SFA record, fill in the senslab specific and SFA specific
1459 fields in the record.
1462 logger.debug("SLABDRIVER \tfill_record_info records %s " %(record_list))
1463 if not isinstance(record_list, list):
1464 record_list = [record_list]
1467 for record in record_list:
1468 #If the record is a SFA slice record, then add information
1469 #about the user of this slice. This kind of
1470 #information is in the Senslab's DB.
1471 if str(record['type']) == 'slice':
1472 if 'reg_researchers' in record and isinstance(record['reg_researchers'],list) :
1473 record['reg_researchers'] = record['reg_researchers'][0].__dict__
1474 record.update({'PI':[record['reg_researchers']['hrn']],
1475 'researcher': [record['reg_researchers']['hrn']],
1476 'name':record['hrn'],
1479 'person_ids':[record['reg_researchers']['record_id']],
1480 'geni_urn':'', #For client_helper.py compatibility
1481 'keys':'', #For client_helper.py compatibility
1482 'key_ids':''}) #For client_helper.py compatibility
1485 #Get slab slice record.
1486 recslice_list = self.GetSlices(slice_filter = \
1487 str(record['hrn']),\
1488 slice_filter_type = 'slice_hrn')
1490 #recuser = recslice_list[0]['reg_researchers']
1491 ##recuser = dbsession.query(RegRecord).filter_by(record_id = \
1492 ##recslice_list[0]['record_id_user']).first()
1494 #record.update({'PI':[recuser['hrn']],
1495 #'researcher': [recuser['hrn']],
1496 #'name':record['hrn'],
1499 #'person_ids':[recslice_list[0]['reg_researchers']['record_id']],
1500 #'geni_urn':'', #For client_helper.py compatibility
1501 #'keys':'', #For client_helper.py compatibility
1502 #'key_ids':''}) #For client_helper.py compatibility
1503 logger.debug("SLABDRIVER \tfill_record_info TYPE SLICE RECUSER record['hrn'] %s ecord['oar_job_id'] %s " %(record['hrn'],record['oar_job_id']))
1505 for rec in recslice_list:
1506 logger.debug("SLABDRIVER\r\n \t \t fill_record_info oar_job_id %s " %(rec['oar_job_id']))
1507 #record['oar_job_id'].append(rec['oar_job_id'])
1508 #del record['_sa_instance_state']
1509 del record['reg_researchers']
1510 record['node_ids'] = [ self.root_auth + hostname for hostname in rec['node_ids']]
1514 logger.debug( "SLABDRIVER.PY \t fill_record_info SLICE \
1515 recslice_list %s \r\n \t RECORD %s \r\n \r\n" %(recslice_list,record))
1516 if str(record['type']) == 'user':
1517 #The record is a SFA user record.
1518 #Get the information about his slice from Senslab's DB
1519 #and add it to the user record.
1520 recslice_list = self.GetSlices(\
1521 slice_filter = record['record_id'],\
1522 slice_filter_type = 'record_id_user')
1524 logger.debug( "SLABDRIVER.PY \t fill_record_info TYPE USER \
1525 recslice_list %s \r\n \t RECORD %s \r\n" %(recslice_list , record))
1526 #Append slice record in records list,
1527 #therefore fetches user and slice info again(one more loop)
1528 #Will update PIs and researcher for the slice
1529 #recuser = dbsession.query(RegRecord).filter_by(record_id = \
1530 #recslice_list[0]['record_id_user']).first()
1531 recuser = recslice_list[0]['reg_researchers']
1532 logger.debug( "SLABDRIVER.PY \t fill_record_info USER \
1533 recuser %s \r\n \r\n" %(recuser))
1535 recslice = recslice_list[0]
1536 recslice.update({'PI':[recuser['hrn']],
1537 'researcher': [recuser['hrn']],
1538 'name':record['hrn'],
1541 'person_ids':[recuser['record_id']]})
1543 for rec in recslice_list:
1544 recslice['oar_job_id'].append(rec['oar_job_id'])
1548 recslice.update({'type':'slice', \
1549 'hrn':recslice_list[0]['hrn']})
1552 #GetPersons takes [] as filters
1553 #user_slab = self.GetPersons([{'hrn':recuser.hrn}])
1554 user_slab = self.GetPersons([record])
1557 record.update(user_slab[0])
1558 #For client_helper.py compatibility
1559 record.update( { 'geni_urn':'',
1562 record_list.append(recslice)
1564 logger.debug("SLABDRIVER.PY \tfill_record_info ADDING SLICE\
1565 INFO TO USER records %s" %(record_list))
1567 logger.debug("SLABDRIVER.PY \tfill_record_info END \
1568 record %s \r\n \r\n " %(record))
1570 except TypeError, error:
1571 logger.log_exc("SLABDRIVER \t fill_record_info EXCEPTION %s"\
1573 #logger.debug("SLABDRIVER.PY \t fill_record_info ENDENDEND ")
1577 #self.fill_record_slab_info(records)
1583 #TODO Update membership? update_membership_list SA 05/07/12
1584 #def update_membership_list(self, oldRecord, record, listName, addFunc, \
1586 ## get a list of the HRNs tht are members of the old and new records
1588 #oldList = oldRecord.get(listName, [])
1591 #newList = record.get(listName, [])
1593 ## if the lists are the same, then we don't have to update anything
1594 #if (oldList == newList):
1597 ## build a list of the new person ids, by looking up each person to get
1601 #records = table.find({'type': 'user', 'hrn': newList})
1602 #for rec in records:
1603 #newIdList.append(rec['pointer'])
1605 ## build a list of the old person ids from the person_ids field
1607 #oldIdList = oldRecord.get("person_ids", [])
1608 #containerId = oldRecord.get_pointer()
1610 ## if oldRecord==None, then we are doing a Register, instead of an
1613 #containerId = record.get_pointer()
1615 ## add people who are in the new list, but not the oldList
1616 #for personId in newIdList:
1617 #if not (personId in oldIdList):
1618 #addFunc(self.plauth, personId, containerId)
1620 ## remove people who are in the old list, but not the new list
1621 #for personId in oldIdList:
1622 #if not (personId in newIdList):
1623 #delFunc(self.plauth, personId, containerId)
1625 #def update_membership(self, oldRecord, record):
1627 #if record.type == "slice":
1628 #self.update_membership_list(oldRecord, record, 'researcher',
1629 #self.users.AddPersonToSlice,
1630 #self.users.DeletePersonFromSlice)
1631 #elif record.type == "authority":
1636 # I don't think you plan on running a component manager at this point
1637 # let me clean up the mess of ComponentAPI that is deprecated anyways
1640 #TODO FUNCTIONS SECTION 04/07/2012 SA
1642 #TODO : Is UnBindObjectFromPeer still necessary ? Currently does nothing
1644 def UnBindObjectFromPeer(self, auth, object_type, object_id, shortname):
1645 """ This method is a hopefully temporary hack to let the sfa correctly
1646 detach the objects it creates from a remote peer object. This is
1647 needed so that the sfa federation link can work in parallel with
1648 RefreshPeer, as RefreshPeer depends on remote objects being correctly
1651 auth : struct, API authentication structure
1652 AuthMethod : string, Authentication method to use
1653 object_type : string, Object type, among 'site','person','slice',
1655 object_id : int, object_id
1656 shortname : string, peer shortname
1660 logger.warning("SLABDRIVER \tUnBindObjectFromPeer EMPTY-\
1664 #TODO Is BindObjectToPeer still necessary ? Currently does nothing
1666 def BindObjectToPeer(self, auth, object_type, object_id, shortname=None, \
1667 remote_object_id=None):
1668 """This method is a hopefully temporary hack to let the sfa correctly
1669 attach the objects it creates to a remote peer object. This is needed
1670 so that the sfa federation link can work in parallel with RefreshPeer,
1671 as RefreshPeer depends on remote objects being correctly marked.
1673 shortname : string, peer shortname
1674 remote_object_id : int, remote object_id, set to 0 if unknown
1678 logger.warning("SLABDRIVER \tBindObjectToPeer EMPTY - DO NOTHING \r\n ")
1681 #TODO UpdateSlice 04/07/2012 SA
1682 #Funciton should delete and create another job since oin senslab slice=job
1683 def UpdateSlice(self, auth, slice_id_or_name, slice_fields=None):
1684 """Updates the parameters of an existing slice with the values in
1686 Users may only update slices of which they are members.
1687 PIs may update any of the slices at their sites, or any slices of
1688 which they are members. Admins may update any slice.
1689 Only PIs and admins may update max_nodes. Slices cannot be renewed
1690 (by updating the expires parameter) more than 8 weeks into the future.
1691 Returns 1 if successful, faults otherwise.
1695 logger.warning("SLABDRIVER UpdateSlice EMPTY - DO NOTHING \r\n ")
1698 #TODO UpdatePerson 04/07/2012 SA
1699 def UpdatePerson(self, slab_hrn, federated_hrn, person_fields=None):
1700 """Updates a person. Only the fields specified in person_fields
1701 are updated, all other fields are left untouched.
1702 Users and techs can only update themselves. PIs can only update
1703 themselves and other non-PIs at their sites.
1704 Returns 1 if successful, faults otherwise.
1708 #new_row = FederatedToSenslab(slab_hrn, federated_hrn)
1709 #slab_dbsession.add(new_row)
1710 #slab_dbsession.commit()
1712 logger.debug("SLABDRIVER UpdatePerson EMPTY - DO NOTHING \r\n ")
1715 #TODO GetKeys 04/07/2012 SA
1716 def GetKeys(self, auth, key_filter=None, return_fields=None):
1717 """Returns an array of structs containing details about keys.
1718 If key_filter is specified and is an array of key identifiers,
1719 or a struct of key attributes, only keys matching the filter
1720 will be returned. If return_fields is specified, only the
1721 specified details will be returned.
1723 Admin may query all keys. Non-admins may only query their own keys.
1727 logger.warning("SLABDRIVER GetKeys EMPTY - DO NOTHING \r\n ")
1730 #TODO DeleteKey 04/07/2012 SA
1731 def DeleteKey(self, auth, key_id):
1733 Non-admins may only delete their own keys.
1734 Returns 1 if successful, faults otherwise.
1738 logger.warning("SLABDRIVER DeleteKey EMPTY - DO NOTHING \r\n ")
1742 #TODO : Check rights to delete person
1743 def DeletePerson(self, auth, person_record):
1744 """ Disable an existing account in senslab LDAP.
1745 Users and techs can only delete themselves. PIs can only
1746 delete themselves and other non-PIs at their sites.
1747 ins can delete anyone.
1748 Returns 1 if successful, faults otherwise.
1752 #Disable user account in senslab LDAP
1753 ret = self.ldap.LdapMarkUserAsDeleted(person_record)
1754 logger.warning("SLABDRIVER DeletePerson %s " %(person_record))
1757 #TODO Check DeleteSlice, check rights 05/07/2012 SA
1758 def DeleteSlice(self, auth, slice_record):
1759 """ Deletes the specified slice.
1760 Senslab : Kill the job associated with the slice if there is one
1761 using DeleteSliceFromNodes.
1762 Updates the slice record in slab db to remove the slice nodes.
1764 Users may only delete slices of which they are members. PIs may
1765 delete any of the slices at their sites, or any slices of which
1766 they are members. Admins may delete any slice.
1767 Returns 1 if successful, faults otherwise.
1771 self.DeleteSliceFromNodes(slice_record)
1772 logger.warning("SLABDRIVER DeleteSlice %s "%(slice_record))
1775 def __add_person_to_db(self, user_dict):
1776 hrn = Xrn(user_dict['urn']).get_hrn()
1777 check_if_exists = dbsession.query(RegUser).filter_by(email = user_dict['email']).first()
1778 #user doesn't exists
1779 if not check_if_exists:
1780 logger.debug("__add_person_to_db \t Adding %s \r\n \r\n \
1781 _________________________________________________________________________\
1783 user_record = RegUser(hrn = hrn, pointer= '-1', authority=get_authority(hrn), \
1784 email= user_dict['email'], gid = None)
1785 user_record.reg_keys = [RegKey(user_dict['pkey'])]
1786 user_record.just_created()
1787 dbsession.add (user_record)
1791 #TODO AddPerson 04/07/2012 SA
1792 #def AddPerson(self, auth, person_fields=None):
1793 def AddPerson(self, record):#TODO fixing 28/08//2012 SA
1794 """Adds a new account. Any fields specified in records are used,
1795 otherwise defaults are used.
1796 Accounts are disabled by default. To enable an account,
1798 Returns the new person_id (> 0) if successful, faults otherwise.
1802 ret = self.ldap.LdapAddUser(record)
1803 logger.debug("SLABDRIVER AddPerson return code %s \r\n "%(ret))
1804 self.__add_person_to_db(record)
1807 #TODO AddPersonToSite 04/07/2012 SA
1808 def AddPersonToSite (self, auth, person_id_or_email, \
1809 site_id_or_login_base=None):
1810 """ Adds the specified person to the specified site. If the person is
1811 already a member of the site, no errors are returned. Does not change
1812 the person's primary site.
1813 Returns 1 if successful, faults otherwise.
1817 logger.warning("SLABDRIVER AddPersonToSite EMPTY - DO NOTHING \r\n ")
1820 #TODO AddRoleToPerson : Not sure if needed in senslab 04/07/2012 SA
1821 def AddRoleToPerson(self, auth, role_id_or_name, person_id_or_email):
1822 """Grants the specified role to the person.
1823 PIs can only grant the tech and user roles to users and techs at their
1824 sites. Admins can grant any role to any user.
1825 Returns 1 if successful, faults otherwise.
1829 logger.warning("SLABDRIVER AddRoleToPerson EMPTY - DO NOTHING \r\n ")
1832 #TODO AddPersonKey 04/07/2012 SA
1833 def AddPersonKey(self, auth, person_id_or_email, key_fields=None):
1834 """Adds a new key to the specified account.
1835 Non-admins can only modify their own keys.
1836 Returns the new key_id (> 0) if successful, faults otherwise.
1840 logger.warning("SLABDRIVER AddPersonKey EMPTY - DO NOTHING \r\n ")
1843 def DeleteLeases(self, leases_id_list, slice_hrn ):
1844 logger.debug("SLABDRIVER DeleteLeases leases_id_list %s slice_hrn %s \
1845 \r\n " %(leases_id_list, slice_hrn))
1846 for job_id in leases_id_list:
1847 self.DeleteJobs(job_id, slice_hrn)