1 from datetime import datetime
3 from sfa.util.sfalogging import logger
5 from sfa.storage.alchemy import dbsession
6 from sqlalchemy.orm import joinedload
7 from sfa.storage.model import RegRecord, RegUser, RegSlice, RegKey
8 # from sfa.iotlab.iotlabpostgres import iotlab_dbsession, IotlabXP
9 from sfa.iotlab.iotlabpostgres import IotlabDB, IotlabXP
10 from sfa.iotlab.OARrestapi import OARrestapi
11 from sfa.iotlab.LDAPapi import LDAPapi
13 from sfa.util.xrn import Xrn, hrn_to_urn, get_authority
15 from sfa.trust.certificate import Keypair, convert_public_key
16 from sfa.trust.gid import create_uuid
17 from sfa.trust.hierarchy import Hierarchy
19 from sfa.iotlab.iotlabaggregate import iotlab_xrn_object
21 class IotlabTestbedAPI():
22 """ Class enabled to use LDAP and OAR api calls. """
24 def __init__(self, config):
25 """Creates an instance of OARrestapi and LDAPapi which will be used to
26 issue calls to OAR or LDAP methods.
27 Set the time format and the testbed granularity used for OAR
28 reservation and leases.
30 :param config: configuration object from sfa.util.config
31 :type config: Config object
33 self.iotlab_db = IotlabDB(config)
34 self.oar = OARrestapi()
36 self.time_format = "%Y-%m-%d %H:%M:%S"
37 self.root_auth = config.SFA_REGISTRY_ROOT_AUTH
38 self.grain = 1 # 10 mins lease minimum
39 #import logging, logging.handlers
40 #from sfa.util.sfalogging import _SfaLogger
41 #sql_logger = _SfaLogger(loggername = 'sqlalchemy.engine', \
47 def GetMinExperimentDurationInSec():
51 def GetPeers (peer_filter=None ):
52 """ Gathers registered authorities in SFA DB and looks for specific peer
53 if peer_filter is specified.
54 :param peer_filter: name of the site authority looked for.
55 :type peer_filter: string
56 :returns: list of records.
61 existing_hrns_by_types = {}
62 logger.debug("IOTLABDRIVER \tGetPeers peer_filter %s, \
64 all_records = dbsession.query(RegRecord).filter(RegRecord.type.like('%authority%')).all()
66 for record in all_records:
67 existing_records[(record.hrn, record.type)] = record
68 if record.type not in existing_hrns_by_types:
69 existing_hrns_by_types[record.type] = [record.hrn]
71 existing_hrns_by_types[record.type].append(record.hrn)
74 logger.debug("IOTLABDRIVER \tGetPeer\texisting_hrns_by_types %s "\
75 %( existing_hrns_by_types))
80 records_list.append(existing_records[(peer_filter,'authority')])
82 for hrn in existing_hrns_by_types['authority']:
83 records_list.append(existing_records[(hrn,'authority')])
85 logger.debug("IOTLABDRIVER \tGetPeer \trecords_list %s " \
91 return_records = records_list
92 logger.debug("IOTLABDRIVER \tGetPeer return_records %s " \
98 #TODO : Handling OR request in make_ldap_filters_from_records
99 #instead of the for loop
100 #over the records' list
101 def GetPersons(self, person_filter=None):
103 Get the enabled users and their properties from Iotlab LDAP.
104 If a filter is specified, looks for the user whose properties match
105 the filter, otherwise returns the whole enabled users'list.
106 :param person_filter: Must be a list of dictionnaries
107 with users properties when not set to None.
108 :param person_filter: list of dict
109 :returns:Returns a list of users whose accounts are enabled
111 :rtype: list of dicts
114 logger.debug("IOTLABDRIVER \tGetPersons person_filter %s" \
117 if person_filter and isinstance(person_filter, list):
118 #If we are looking for a list of users (list of dict records)
119 #Usually the list contains only one user record
120 for searched_attributes in person_filter:
122 #Get only enabled user accounts in iotlab LDAP :
123 #add a filter for make_ldap_filters_from_record
124 person = self.ldap.LdapFindUser(searched_attributes, \
125 is_user_enabled=True)
126 #If a person was found, append it to the list
128 person_list.append(person)
130 #If the list is empty, return None
131 if len(person_list) is 0:
135 #Get only enabled user accounts in iotlab LDAP :
136 #add a filter for make_ldap_filters_from_record
137 person_list = self.ldap.LdapFindUser(is_user_enabled=True)
142 #def GetTimezone(self):
143 #""" Returns the OAR server time and timezone.
144 #Unused SA 30/05/13"""
145 #server_timestamp, server_tz = self.oar.parser.\
146 #SendRequest("GET_timezone")
147 #return server_timestamp, server_tz
149 def DeleteJobs(self, job_id, username):
151 """ Deletes the job with the specified job_id and username on OAR by
152 posting a delete request to OAR.
154 :param job_id: job id in OAR.
155 :param username: user's iotlab login in LDAP.
157 :type username: string
159 :returns: dictionary with the job id and if delete has been successful
163 logger.debug("IOTLABDRIVER \tDeleteJobs jobid %s username %s "\
165 if not job_id or job_id is -1:
169 reqdict['method'] = "delete"
170 reqdict['strval'] = str(job_id)
173 answer = self.oar.POSTRequestToOARRestAPI('DELETE_jobs_id', \
175 if answer['status'] == 'Delete request registered':
176 ret = {job_id : True }
178 ret = {job_id :False }
179 logger.debug("IOTLABDRIVER \tDeleteJobs jobid %s \r\n answer %s \
180 username %s" %(job_id, answer, username))
185 ##TODO : Unused GetJobsId ? SA 05/07/12
186 #def GetJobsId(self, job_id, username = None ):
188 #Details about a specific job.
189 #Includes details about submission time, jot type, state, events,
190 #owner, assigned ressources, walltime etc...
194 #node_list_k = 'assigned_network_address'
195 ##Get job info from OAR
196 #job_info = self.oar.parser.SendRequest(req, job_id, username)
198 #logger.debug("IOTLABDRIVER \t GetJobsId %s " %(job_info))
200 #if job_info['state'] == 'Terminated':
201 #logger.debug("IOTLABDRIVER \t GetJobsId job %s TERMINATED"\
204 #if job_info['state'] == 'Error':
205 #logger.debug("IOTLABDRIVER \t GetJobsId ERROR message %s "\
210 #logger.error("IOTLABDRIVER \tGetJobsId KeyError")
213 #parsed_job_info = self.get_info_on_reserved_nodes(job_info, \
215 ##Replaces the previous entry
216 ##"assigned_network_address" / "reserved_resources"
218 #job_info.update({'node_ids':parsed_job_info[node_list_k]})
219 #del job_info[node_list_k]
220 #logger.debug(" \r\nIOTLABDRIVER \t GetJobsId job_info %s " %(job_info))
224 def GetJobsResources(self, job_id, username = None):
225 """ Gets the list of nodes associated with the job_id and username
227 Transforms the iotlab hostnames to the corresponding
229 Rertuns dict key :'node_ids' , value : hostnames list
230 :param username: user's LDAP login
231 :paran job_id: job's OAR identifier.
232 :type username: string
233 :type job_id: integer
235 :returns: dicionary with nodes' hostnames belonging to the job.
239 req = "GET_jobs_id_resources"
242 #Get job resources list from OAR
243 node_id_list = self.oar.parser.SendRequest(req, job_id, username)
244 logger.debug("IOTLABDRIVER \t GetJobsResources %s " %(node_id_list))
247 self.__get_hostnames_from_oar_node_ids(node_id_list)
250 #Replaces the previous entry "assigned_network_address" /
251 #"reserved_resources" with "node_ids"
252 job_info = {'node_ids': hostname_list}
257 #def get_info_on_reserved_nodes(self, job_info, node_list_name):
259 #..warning:unused SA 23/05/13
261 ##Get the list of the testbed nodes records and make a
262 ##dictionnary keyed on the hostname out of it
263 #node_list_dict = self.GetNodes()
264 ##node_hostname_list = []
265 #node_hostname_list = [node['hostname'] for node in node_list_dict]
266 ##for node in node_list_dict:
267 ##node_hostname_list.append(node['hostname'])
268 #node_dict = dict(zip(node_hostname_list, node_list_dict))
270 #reserved_node_hostname_list = []
271 #for index in range(len(job_info[node_list_name])):
272 ##job_info[node_list_name][k] =
273 #reserved_node_hostname_list[index] = \
274 #node_dict[job_info[node_list_name][index]]['hostname']
276 #logger.debug("IOTLABDRIVER \t get_info_on_reserved_nodes \
277 #reserved_node_hostname_list %s" \
278 #%(reserved_node_hostname_list))
280 #logger.error("IOTLABDRIVER \t get_info_on_reserved_nodes KEYERROR " )
282 #return reserved_node_hostname_list
284 def GetNodesCurrentlyInUse(self):
285 """Returns a list of all the nodes already involved in an oar running
287 :rtype: list of nodes hostnames.
289 return self.oar.parser.SendRequest("GET_running_jobs")
291 def __get_hostnames_from_oar_node_ids(self, resource_id_list ):
292 """Get the hostnames of the nodes from their OAR identifiers.
293 Get the list of nodes dict using GetNodes and find the hostname
294 associated with the identifier.
295 :param resource_id_list: list of nodes identifiers
296 :returns: list of node hostnames.
298 full_nodes_dict_list = self.GetNodes()
299 #Put the full node list into a dictionary keyed by oar node id
300 oar_id_node_dict = {}
301 for node in full_nodes_dict_list:
302 oar_id_node_dict[node['oar_id']] = node
305 for resource_id in resource_id_list:
306 #Because jobs requested "asap" do not have defined resources
307 if resource_id is not "Undefined":
308 hostname_list.append(\
309 oar_id_node_dict[resource_id]['hostname'])
311 #hostname_list.append(oar_id_node_dict[resource_id]['hostname'])
314 def GetReservedNodes(self, username = None):
315 """ Get list of leases. Get the leases for the username if specified,
316 otherwise get all the leases. Finds the nodes hostnames for each
318 :param username: user's LDAP login
319 :type username: string
320 :returns: list of reservations dict
324 #Get the nodes in use and the reserved nodes
325 reservation_dict_list = \
326 self.oar.parser.SendRequest("GET_reserved_nodes", \
330 for resa in reservation_dict_list:
331 logger.debug ("GetReservedNodes resa %s"%(resa))
332 #dict list of hostnames and their site
333 resa['reserved_nodes'] = \
334 self.__get_hostnames_from_oar_node_ids(resa['resource_ids'])
336 #del resa['resource_ids']
337 return reservation_dict_list
339 def GetNodes(self, node_filter_dict = None, return_fields_list = None):
342 Make a list of iotlab nodes and their properties from information
343 given by OAR. Search for specific nodes if some filters are specified.
344 Nodes properties returned if no return_fields_list given:
345 'hrn','archi','mobile','hostname','site','boot_state','node_id',
346 'radio','posx','posy','oar_id','posz'.
348 :param node_filter_dict: dictionnary of lists with node properties
349 :type node_filter_dict: dict
350 :param return_fields_list: list of specific fields the user wants to be
352 :type return_fields_list: list
353 :returns: list of dictionaries with node properties
357 node_dict_by_id = self.oar.parser.SendRequest("GET_resources_full")
358 node_dict_list = node_dict_by_id.values()
359 logger.debug (" IOTLABDRIVER GetNodes node_filter_dict %s \
360 return_fields_list %s "%(node_filter_dict, return_fields_list))
361 #No filtering needed return the list directly
362 if not (node_filter_dict or return_fields_list):
363 return node_dict_list
365 return_node_list = []
367 for filter_key in node_filter_dict:
369 #Filter the node_dict_list by each value contained in the
370 #list node_filter_dict[filter_key]
371 for value in node_filter_dict[filter_key]:
372 for node in node_dict_list:
373 if node[filter_key] == value:
374 if return_fields_list :
376 for k in return_fields_list:
378 return_node_list.append(tmp)
380 return_node_list.append(node)
382 logger.log_exc("GetNodes KeyError")
386 return return_node_list
391 def AddSlice(slice_record, user_record):
392 """Add slice to the local iotlab sfa tables if the slice comes
393 from a federated site and is not yet in the iotlab sfa DB,
394 although the user has already a LDAP login.
395 Called by verify_slice during lease/sliver creation.
396 :param slice_record: record of slice, must contain hrn, gid, slice_id
397 and authority of the slice.
398 :type slice_record: dictionary
399 :param user_record: record of the user
400 :type user_record: RegUser
403 sfa_record = RegSlice(hrn=slice_record['hrn'],
404 gid=slice_record['gid'],
405 pointer=slice_record['slice_id'],
406 authority=slice_record['authority'])
407 logger.debug("IOTLABDRIVER.PY AddSlice sfa_record %s user_record %s"
408 % (sfa_record, user_record))
409 sfa_record.just_created()
410 dbsession.add(sfa_record)
412 #Update the reg-researcher dependance table
413 sfa_record.reg_researchers = [user_record]
419 def GetSites(self, site_filter_name_list = None, return_fields_list = None):
420 site_dict = self.oar.parser.SendRequest("GET_sites")
421 #site_dict : dict where the key is the sit ename
422 return_site_list = []
423 if not ( site_filter_name_list or return_fields_list):
424 return_site_list = site_dict.values()
425 return return_site_list
427 for site_filter_name in site_filter_name_list:
428 if site_filter_name in site_dict:
429 if return_fields_list:
430 for field in return_fields_list:
433 tmp[field] = site_dict[site_filter_name][field]
435 logger.error("GetSites KeyError %s "%(field))
437 return_site_list.append(tmp)
439 return_site_list.append( site_dict[site_filter_name])
442 return return_site_list
448 #TODO : Check rights to delete person
449 def DeletePerson(self, person_record):
450 """ Disable an existing account in iotlab LDAP.
451 Users and techs can only delete themselves. PIs can only
452 delete themselves and other non-PIs at their sites.
453 ins can delete anyone.
454 :param person_record: user's record
455 :type person_record: dict
456 :returns: True if successful, False otherwise.
460 #Disable user account in iotlab LDAP
461 ret = self.ldap.LdapMarkUserAsDeleted(person_record)
462 logger.warning("IOTLABDRIVER DeletePerson %s " %(person_record))
466 def DeleteSlice(self, slice_record):
467 """ Deletes the specified slice and kills the jobs associated with
468 the slice if any, using DeleteSliceFromNodes.
470 :returns: True if all the jobs in the slice have been deleted,
471 or the list of jobs that could not be deleted otherwise.
472 :rtype: list or boolean
475 ret = self.DeleteSliceFromNodes(slice_record)
478 if False in ret[job_id]:
479 if delete_failed is None:
481 delete_failed.append(job_id)
483 logger.info("IOTLABDRIVER DeleteSlice %s answer %s"%(slice_record, \
485 return delete_failed or True
488 def __add_person_to_db(user_dict):
490 Add a federated user straight to db when the user issues a lease
491 request with iotlab nodes and that he has not registered with iotlab
492 yet (that is he does not have a LDAP entry yet).
493 Uses parts of the routines in SlabImport when importing user from LDAP.
494 Called by AddPerson, right after LdapAddUser.
495 :param user_dict: Must contain email, hrn and pkey to get a GID
496 and be added to the SFA db.
497 :type user_dict: dict
501 dbsession.query(RegUser).filter_by(email = user_dict['email']).first()
503 if not check_if_exists:
504 logger.debug("__add_person_to_db \t Adding %s \r\n \r\n \
506 hrn = user_dict['hrn']
507 person_urn = hrn_to_urn(hrn, 'user')
508 pubkey = user_dict['pkey']
510 pkey = convert_public_key(pubkey)
512 #key not good. create another pkey
513 logger.warn('__add_person_to_db: unable to convert public \
515 pkey = Keypair(create=True)
518 if pubkey is not None and pkey is not None :
519 hierarchy = Hierarchy()
520 person_gid = hierarchy.create_gid(person_urn, create_uuid(), \
522 if user_dict['email']:
523 logger.debug("__add_person_to_db \r\n \r\n \
524 IOTLAB IMPORTER PERSON EMAIL OK email %s "\
525 %(user_dict['email']))
526 person_gid.set_email(user_dict['email'])
528 user_record = RegUser(hrn=hrn , pointer= '-1', \
529 authority=get_authority(hrn), \
530 email=user_dict['email'], gid = person_gid)
531 user_record.reg_keys = [RegKey(user_dict['pkey'])]
532 user_record.just_created()
533 dbsession.add (user_record)
538 def AddPerson(self, record):
539 """Adds a new account. Any fields specified in records are used,
540 otherwise defaults are used. Creates an appropriate login by calling
542 :param record: dictionary with the sfa user's properties.
543 :returns: The uid of the added person if sucessful, otherwise returns
544 the error message from LDAP.
545 :rtype: interger or string
547 ret = self.ldap.LdapAddUser(record)
549 if ret['bool'] is True:
550 record['hrn'] = self.root_auth + '.' + ret['uid']
551 logger.debug("IOTLABDRIVER AddPerson return code %s record %s \r\n "\
553 self.__add_person_to_db(record)
556 return ret['message']
560 #TODO AddPersonKey 04/07/2012 SA
561 def AddPersonKey(self, person_uid, old_attributes_dict, new_key_dict):
562 """Adds a new key to the specified account. Adds the key to the
563 iotlab ldap, provided that the person_uid is valid.
564 Non-admins can only modify their own keys.
566 :param person_uid: user's iotlab login in LDAP
567 :param old_attributes_dict: dict with the user's old sshPublicKey
568 :param new_key_dict:dict with the user's new sshPublicKey
569 :type person_uid: string
573 :returns: True if the key has been modified, False otherwise.
576 ret = self.ldap.LdapModify(person_uid, old_attributes_dict, \
578 logger.warning("IOTLABDRIVER AddPersonKey EMPTY - DO NOTHING \r\n ")
581 def DeleteLeases(self, leases_id_list, slice_hrn ):
583 Deletes several leases, based on their job ids and the slice
584 they are associated with. Uses DeleteJobs to delete the jobs
585 on OAR. Note that one slice can contain multiple jobs, and in this case
586 all the jobs in the leases_id_list MUST belong to ONE slice,
587 since there is only one slice hrn provided here.
588 :param leases_id_list: list of job ids that belong to the slice whose
589 slice hrn is provided.
590 :param slice_hrn: the slice hrn .
591 ..warning: Does not have a return value since there was no easy
592 way to handle failure when dealing with multiple job delete. Plus,
593 there was no easy way to report it to the user.
595 logger.debug("IOTLABDRIVER DeleteLeases leases_id_list %s slice_hrn %s \
596 \r\n " %(leases_id_list, slice_hrn))
597 for job_id in leases_id_list:
598 self.DeleteJobs(job_id, slice_hrn)
603 def _process_walltime(duration):
604 """ Calculates the walltime in seconds from the duration in H:M:S
605 specified in the RSpec.
609 # Fixing the walltime by adding a few delays.
610 # First put the walltime in seconds oarAdditionalDelay = 20;
611 # additional delay for /bin/sleep command to
612 # take in account prologue and epilogue scripts execution
613 # int walltimeAdditionalDelay = 240; additional delay
614 #for prologue/epilogue execution = $SERVER_PROLOGUE_EPILOGUE_TIMEOUT
616 # Put the duration in seconds first
617 #desired_walltime = duration * 60
618 desired_walltime = duration
619 total_walltime = desired_walltime + 240 #+4 min Update SA 23/10/12
620 sleep_walltime = desired_walltime # 0 sec added Update SA 23/10/12
622 #Put the walltime back in str form
624 walltime.append(str(total_walltime / 3600))
625 total_walltime = total_walltime - 3600 * int(walltime[0])
626 #Get the remaining minutes
627 walltime.append(str(total_walltime / 60))
628 total_walltime = total_walltime - 60 * int(walltime[1])
630 walltime.append(str(total_walltime))
633 logger.log_exc(" __process_walltime duration null")
635 return walltime, sleep_walltime
638 def _create_job_structure_request_for_OAR(lease_dict):
639 """ Creates the structure needed for a correct POST on OAR.
640 Makes the timestamp transformation into the appropriate format.
641 Sends the POST request to create the job with the resources in
650 reqdict['workdir'] = '/tmp'
651 reqdict['resource'] = "{network_address in ("
653 for node in lease_dict['added_nodes']:
654 logger.debug("\r\n \r\n OARrestapi \t \
655 __create_job_structure_request_for_OAR node %s" %(node))
657 # Get the ID of the node
659 reqdict['resource'] += "'" + nodeid + "', "
660 nodeid_list.append(nodeid)
662 custom_length = len(reqdict['resource'])- 2
663 reqdict['resource'] = reqdict['resource'][0:custom_length] + \
664 ")}/nodes=" + str(len(nodeid_list))
667 walltime, sleep_walltime = \
668 IotlabTestbedAPI._process_walltime(\
669 int(lease_dict['lease_duration']))
672 reqdict['resource'] += ",walltime=" + str(walltime[0]) + \
673 ":" + str(walltime[1]) + ":" + str(walltime[2])
674 reqdict['script_path'] = "/bin/sleep " + str(sleep_walltime)
676 #In case of a scheduled experiment (not immediate)
677 #To run an XP immediately, don't specify date and time in RSpec
678 #They will be set to None.
679 if lease_dict['lease_start_time'] is not '0':
680 #Readable time accepted by OAR
681 start_time = datetime.fromtimestamp( \
682 int(lease_dict['lease_start_time'])).\
683 strftime(lease_dict['time_format'])
684 reqdict['reservation'] = start_time
685 #If there is not start time, Immediate XP. No need to add special
689 reqdict['type'] = "deploy"
690 reqdict['directory'] = ""
691 reqdict['name'] = "SFA_" + lease_dict['slice_user']
696 def LaunchExperimentOnOAR(self, added_nodes, slice_name, \
697 lease_start_time, lease_duration, slice_user=None):
700 Create a job request structure based on the information provided
701 and post the job on OAR.
702 :param added_nodes: list of nodes that belong to the described lease.
703 :param slice_name: the slice hrn associated to the lease.
704 :param lease_start_time: timestamp of the lease startting time.
705 :param lease_duration: lease durationin minutes
709 lease_dict['lease_start_time'] = lease_start_time
710 lease_dict['lease_duration'] = lease_duration
711 lease_dict['added_nodes'] = added_nodes
712 lease_dict['slice_name'] = slice_name
713 lease_dict['slice_user'] = slice_user
714 lease_dict['grain'] = self.GetLeaseGranularity()
715 lease_dict['time_format'] = self.time_format
718 logger.debug("IOTLABDRIVER.PY \tLaunchExperimentOnOAR slice_user %s\
719 \r\n " %(slice_user))
720 #Create the request for OAR
721 reqdict = self._create_job_structure_request_for_OAR(lease_dict)
722 # first step : start the OAR job and update the job
723 logger.debug("IOTLABDRIVER.PY \tLaunchExperimentOnOAR reqdict %s\
726 answer = self.oar.POSTRequestToOARRestAPI('POST_job', \
728 logger.debug("IOTLABDRIVER \tLaunchExperimentOnOAR jobid %s " %(answer))
732 logger.log_exc("IOTLABDRIVER \tLaunchExperimentOnOAR \
733 Impossible to create job %s " %(answer))
740 logger.debug("IOTLABDRIVER \tLaunchExperimentOnOAR jobid %s \
741 added_nodes %s slice_user %s" %(jobid, added_nodes, \
748 def AddLeases(self, hostname_list, slice_record, \
749 lease_start_time, lease_duration):
751 """Creates a job in OAR corresponding to the information provided
752 as parameters. Adds the job id and the slice hrn in the iotlab
753 database so that we are able to know which slice has which nodes.
755 :param hostname_list: list of nodes' OAR hostnames.
756 :param slice_record: sfa slice record, must contain login and hrn.
757 :param lease_start_time: starting time , unix timestamp format
758 :param lease_duration: duration in minutes
760 :type hostname_list: list
761 :type slice_record: dict
762 :type lease_start_time: integer
763 :type lease_duration: integer
766 logger.debug("IOTLABDRIVER \r\n \r\n \t AddLeases hostname_list %s \
767 slice_record %s lease_start_time %s lease_duration %s "\
768 %( hostname_list, slice_record , lease_start_time, \
771 #tmp = slice_record['reg-researchers'][0].split(".")
772 username = slice_record['login']
773 #username = tmp[(len(tmp)-1)]
774 job_id = self.LaunchExperimentOnOAR(hostname_list, \
775 slice_record['hrn'], \
776 lease_start_time, lease_duration, \
779 datetime.fromtimestamp(int(lease_start_time)).\
780 strftime(self.time_format)
781 end_time = lease_start_time + lease_duration
784 logger.debug("IOTLABDRIVER \r\n \r\n \t AddLeases TURN ON LOGGING SQL \
785 %s %s %s "%(slice_record['hrn'], job_id, end_time))
788 logger.debug("IOTLABDRIVER \r\n \r\n \t AddLeases %s %s %s " \
789 %(type(slice_record['hrn']), type(job_id), type(end_time)))
791 iotlab_ex_row = IotlabXP(slice_hrn = slice_record['hrn'], \
792 job_id = job_id, end_time= end_time)
794 logger.debug("IOTLABDRIVER \r\n \r\n \t AddLeases iotlab_ex_row %s" \
796 self.iotlab_db.iotlab_session.add(iotlab_ex_row)
797 self.iotlab_db.iotlab_session.commit()
799 logger.debug("IOTLABDRIVER \t AddLeases hostname_list start_time %s " \
805 #Delete the jobs from job_iotlab table
806 def DeleteSliceFromNodes(self, slice_record):
807 """ Deletes all the running or scheduled jobs of a given slice
809 :param slice_record: record of the slice
810 :type slice_record: dict
812 :returns: dict of the jobs'deletion status. Success= True, Failure=
813 False, for each job id.
816 logger.debug("IOTLABDRIVER \t DeleteSliceFromNodese %s " %(slice_record))
818 if isinstance(slice_record['oar_job_id'], list):
820 for job_id in slice_record['oar_job_id']:
821 ret = self.DeleteJobs(job_id, slice_record['user'])
823 oar_bool_answer.update(ret)
826 oar_bool_answer = [self.DeleteJobs(slice_record['oar_job_id'], \
827 slice_record['user'])]
829 return oar_bool_answer
833 def GetLeaseGranularity(self):
834 """ Returns the granularity of an experiment in the Iotlab testbed.
835 OAR uses seconds for experiments duration , the granulaity is also
837 Experiments which last less than 10 min (600 sec) are invalid"""
842 # def update_jobs_in_iotlabdb( job_oar_list, jobs_psql):
843 # """ Cleans the iotlab db by deleting expired and cancelled jobs.
844 # Compares the list of job ids given by OAR with the job ids that
845 # are already in the database, deletes the jobs that are no longer in
846 # the OAR job id list.
847 # :param job_oar_list: list of job ids coming from OAR
848 # :type job_oar_list: list
849 # :param job_psql: list of job ids cfrom the database.
850 # type job_psql: list
852 # #Turn the list into a set
853 # set_jobs_psql = set(jobs_psql)
855 # kept_jobs = set(job_oar_list).intersection(set_jobs_psql)
856 # logger.debug ( "\r\n \t\ update_jobs_in_iotlabdb jobs_psql %s \r\n \t \
857 # job_oar_list %s kept_jobs %s "%(set_jobs_psql, job_oar_list, kept_jobs))
858 # deleted_jobs = set_jobs_psql.difference(kept_jobs)
859 # deleted_jobs = list(deleted_jobs)
860 # if len(deleted_jobs) > 0:
861 # self.iotlab_db.iotlab_session.query(IotlabXP).filter(IotlabXP.job_id.in_(deleted_jobs)).delete(synchronize_session='fetch')
862 # self.iotlab_db.iotlab_session.commit()
868 def GetLeases(self, lease_filter_dict=None, login=None):
869 """ Get the list of leases from OAR with complete information
870 about which slice owns which jobs and nodes.
872 -Fetch all the jobs from OAR (running, waiting..)
873 complete the reservation information with slice hrn
874 found in iotlab_xp table. If not available in the table,
875 assume it is a iotlab slice.
876 -Updates the iotlab table, deleting jobs when necessary.
877 :returns: reservation_list, list of dictionaries with 'lease_id',
878 'reserved_nodes','slice_id', 'state', 'user', 'component_id_list',
879 'slice_hrn', 'resource_ids', 't_from', 't_until'
883 unfiltered_reservation_list = self.GetReservedNodes(login)
885 reservation_list = []
886 #Find the slice associated with this user iotlab ldap uid
887 logger.debug(" IOTLABDRIVER.PY \tGetLeases login %s\
888 unfiltered_reservation_list %s " %(login, unfiltered_reservation_list))
889 #Create user dict first to avoid looking several times for
890 #the same user in LDAP SA 27/07/12
893 jobs_psql_query = self.iotlab_db.iotlab_session.query(IotlabXP).all()
894 jobs_psql_dict = dict([(row.job_id, row.__dict__ ) for row in jobs_psql_query ])
895 #jobs_psql_dict = jobs_psql_dict)
896 logger.debug("IOTLABDRIVER \tGetLeases jobs_psql_dict %s"\
898 jobs_psql_id_list = [ row.job_id for row in jobs_psql_query ]
902 for resa in unfiltered_reservation_list:
903 logger.debug("IOTLABDRIVER \tGetLeases USER %s"\
905 #Construct list of jobs (runing, waiting..) in oar
906 job_oar_list.append(resa['lease_id'])
907 #If there is information on the job in IOTLAB DB ]
908 #(slice used and job id)
909 if resa['lease_id'] in jobs_psql_dict:
910 job_info = jobs_psql_dict[resa['lease_id']]
911 logger.debug("IOTLABDRIVER \tGetLeases job_info %s"\
913 resa['slice_hrn'] = job_info['slice_hrn']
914 resa['slice_id'] = hrn_to_urn(resa['slice_hrn'], 'slice')
916 #otherwise, assume it is a iotlab slice:
918 resa['slice_id'] = hrn_to_urn(self.root_auth+'.'+ \
919 resa['user'] +"_slice" , 'slice')
920 resa['slice_hrn'] = Xrn(resa['slice_id']).get_hrn()
922 resa['component_id_list'] = []
923 #Transform the hostnames into urns (component ids)
924 for node in resa['reserved_nodes']:
926 iotlab_xrn = iotlab_xrn_object(self.root_auth, node)
927 resa['component_id_list'].append(iotlab_xrn.urn)
929 if lease_filter_dict:
930 logger.debug("IOTLABDRIVER \tGetLeases resa_ %s \
931 \r\n leasefilter %s" %(resa, lease_filter_dict))
933 if lease_filter_dict['name'] == resa['slice_hrn']:
934 reservation_list.append(resa)
936 if lease_filter_dict is None:
937 reservation_list = unfiltered_reservation_list
940 self.update_jobs_in_iotlabdb(job_oar_list, jobs_psql_id_list)
942 logger.debug(" IOTLABDRIVER.PY \tGetLeases reservation_list %s"\
944 return reservation_list
949 #TODO FUNCTIONS SECTION 04/07/2012 SA
951 ##TODO : Is UnBindObjectFromPeer still necessary ? Currently does nothing
954 #def UnBindObjectFromPeer( auth, object_type, object_id, shortname):
955 #""" This method is a hopefully temporary hack to let the sfa correctly
956 #detach the objects it creates from a remote peer object. This is
957 #needed so that the sfa federation link can work in parallel with
958 #RefreshPeer, as RefreshPeer depends on remote objects being correctly
961 #auth : struct, API authentication structure
962 #AuthMethod : string, Authentication method to use
963 #object_type : string, Object type, among 'site','person','slice',
965 #object_id : int, object_id
966 #shortname : string, peer shortname
970 #logger.warning("IOTLABDRIVER \tUnBindObjectFromPeer EMPTY-\
974 ##TODO Is BindObjectToPeer still necessary ? Currently does nothing
976 #|| Commented out 28/05/13 SA
977 #def BindObjectToPeer(self, auth, object_type, object_id, shortname=None, \
978 #remote_object_id=None):
979 #"""This method is a hopefully temporary hack to let the sfa correctly
980 #attach the objects it creates to a remote peer object. This is needed
981 #so that the sfa federation link can work in parallel with RefreshPeer,
982 #as RefreshPeer depends on remote objects being correctly marked.
984 #shortname : string, peer shortname
985 #remote_object_id : int, remote object_id, set to 0 if unknown
989 #logger.warning("IOTLABDRIVER \tBindObjectToPeer EMPTY - DO NOTHING \r\n ")
992 ##TODO UpdateSlice 04/07/2012 SA || Commented out 28/05/13 SA
993 ##Funciton should delete and create another job since oin iotlab slice=job
994 #def UpdateSlice(self, auth, slice_id_or_name, slice_fields=None):
995 #"""Updates the parameters of an existing slice with the values in
997 #Users may only update slices of which they are members.
998 #PIs may update any of the slices at their sites, or any slices of
999 #which they are members. Admins may update any slice.
1000 #Only PIs and admins may update max_nodes. Slices cannot be renewed
1001 #(by updating the expires parameter) more than 8 weeks into the future.
1002 #Returns 1 if successful, faults otherwise.
1006 #logger.warning("IOTLABDRIVER UpdateSlice EMPTY - DO NOTHING \r\n ")
1009 #Unused SA 30/05/13, we only update the user's key or we delete it.
1010 ##TODO UpdatePerson 04/07/2012 SA
1011 #def UpdatePerson(self, iotlab_hrn, federated_hrn, person_fields=None):
1012 #"""Updates a person. Only the fields specified in person_fields
1013 #are updated, all other fields are left untouched.
1014 #Users and techs can only update themselves. PIs can only update
1015 #themselves and other non-PIs at their sites.
1016 #Returns 1 if successful, faults otherwise.
1020 ##new_row = FederatedToIotlab(iotlab_hrn, federated_hrn)
1021 ##self.iotlab_db.iotlab_session.add(new_row)
1022 ##self.iotlab_db.iotlab_session.commit()
1024 #logger.debug("IOTLABDRIVER UpdatePerson EMPTY - DO NOTHING \r\n ")
1028 def GetKeys(key_filter=None):
1029 """Returns a dict of dict based on the key string. Each dict entry
1030 contains the key id, the ssh key, the user's email and the
1032 If key_filter is specified and is an array of key identifiers,
1033 only keys matching the filter will be returned.
1035 Admin may query all keys. Non-admins may only query their own keys.
1038 :returns: dict with ssh key as key and dicts as value.
1041 if key_filter is None:
1042 keys = dbsession.query(RegKey).options(joinedload('reg_user')).all()
1044 keys = dbsession.query(RegKey).options(joinedload('reg_user')).filter(RegKey.key.in_(key_filter)).all()
1048 key_dict[key.key] = {'key_id': key.key_id, 'key': key.key, \
1049 'email': key.reg_user.email, 'hrn':key.reg_user.hrn}
1051 #ldap_rslt = self.ldap.LdapSearch({'enabled']=True})
1052 #user_by_email = dict((user[1]['mail'][0], user[1]['sshPublicKey']) \
1053 #for user in ldap_rslt)
1055 logger.debug("IOTLABDRIVER GetKeys -key_dict %s \r\n " %(key_dict))
1059 def DeleteKey(self, user_record, key_string):
1060 """ Deletes a key in the LDAP entry of the specified user.
1061 Removes the key_string from the user's key list and updates the LDAP
1062 user's entry with the new key attributes.
1063 :param key_string: The ssh key to remove
1064 :param user_record: User's record
1065 :type key_string: string
1066 :type user_record: dict
1067 :returns: True if sucessful, False if not.
1071 all_user_keys = user_record['keys']
1072 all_user_keys.remove(key_string)
1073 new_attributes = {'sshPublicKey':all_user_keys}
1074 ret = self.ldap.LdapModifyUser(user_record, new_attributes)
1075 logger.debug("IOTLABDRIVER DeleteKey %s- "%(ret))
1082 def _sql_get_slice_info( slice_filter ):
1084 Get the slice record based on the slice hrn. Fetch the record of the
1085 user associated with the slice by usingjoinedload based on t
1086 he reg_researcher relationship.
1087 :param slice_filter: the slice hrn we are looking for
1088 :type slice_filter: string
1089 :returns: the slice record enhanced with the user's information if the
1090 slice was found, None it wasn't.
1091 :rtype: dict or None.
1093 #DO NOT USE RegSlice - reg_researchers to get the hrn
1094 #of the user otherwise will mess up the RegRecord in
1095 #Resolve, don't know why - SA 08/08/2012
1097 #Only one entry for one user = one slice in iotlab_xp table
1098 #slicerec = dbsession.query(RegRecord).filter_by(hrn = slice_filter).first()
1099 raw_slicerec = dbsession.query(RegSlice).options(joinedload('reg_researchers')).filter_by(hrn = slice_filter).first()
1100 #raw_slicerec = dbsession.query(RegRecord).filter_by(hrn = slice_filter).first()
1102 #load_reg_researcher
1103 #raw_slicerec.reg_researchers
1104 raw_slicerec = raw_slicerec.__dict__
1105 logger.debug(" IOTLABDRIVER \t get_slice_info slice_filter %s \
1106 raw_slicerec %s"%(slice_filter, raw_slicerec))
1107 slicerec = raw_slicerec
1108 #only one researcher per slice so take the first one
1109 #slicerec['reg_researchers'] = raw_slicerec['reg_researchers']
1110 #del slicerec['reg_researchers']['_sa_instance_state']
1117 def _sql_get_slice_info_from_user(slice_filter ):
1119 Get the slice record based on the user recordid by using a joinedload
1120 on the relationship reg_slices_as_researcher. Format the sql record
1121 into a dict with the mandatory fields for user and slice.
1122 :returns: dict with slice record and user record if the record was found
1123 based on the user's id, None if not..
1124 :rtype:dict or None..
1126 #slicerec = dbsession.query(RegRecord).filter_by(record_id = slice_filter).first()
1127 raw_slicerec = dbsession.query(RegUser).options(joinedload('reg_slices_as_researcher')).filter_by(record_id = slice_filter).first()
1128 #raw_slicerec = dbsession.query(RegRecord).filter_by(record_id = slice_filter).first()
1129 #Put it in correct order
1130 user_needed_fields = ['peer_authority', 'hrn', 'last_updated', 'classtype', 'authority', 'gid', 'record_id', 'date_created', 'type', 'email', 'pointer']
1131 slice_needed_fields = ['peer_authority', 'hrn', 'last_updated', 'classtype', 'authority', 'gid', 'record_id', 'date_created', 'type', 'pointer']
1133 #raw_slicerec.reg_slices_as_researcher
1134 raw_slicerec = raw_slicerec.__dict__
1137 dict([(k, raw_slicerec['reg_slices_as_researcher'][0].__dict__[k]) \
1138 for k in slice_needed_fields])
1139 slicerec['reg_researchers'] = dict([(k, raw_slicerec[k]) \
1140 for k in user_needed_fields])
1141 #TODO Handle multiple slices for one user SA 10/12/12
1142 #for now only take the first slice record associated to the rec user
1143 ##slicerec = raw_slicerec['reg_slices_as_researcher'][0].__dict__
1144 #del raw_slicerec['reg_slices_as_researcher']
1145 #slicerec['reg_researchers'] = raw_slicerec
1146 ##del slicerec['_sa_instance_state']
1153 def _get_slice_records(self, slice_filter = None, \
1154 slice_filter_type = None):
1156 Get the slice record depending on the slice filter and its type.
1157 :param slice_filter: Can be either the slice hrn or the user's record
1159 :type slice_filter: string
1160 :param slice_filter_type: describes the slice filter type used, can be
1161 slice_hrn or record_id_user
1163 :returns: the slice record
1165 .. seealso::_sql_get_slice_info_from_user
1166 .. seealso:: _sql_get_slice_info
1169 #Get list of slices based on the slice hrn
1170 if slice_filter_type == 'slice_hrn':
1172 #if get_authority(slice_filter) == self.root_auth:
1173 #login = slice_filter.split(".")[1].split("_")[0]
1175 slicerec = self._sql_get_slice_info(slice_filter)
1177 if slicerec is None:
1181 #Get slice based on user id
1182 if slice_filter_type == 'record_id_user':
1184 slicerec = self._sql_get_slice_info_from_user(slice_filter)
1187 fixed_slicerec_dict = slicerec
1188 #At this point if there is no login it means
1189 #record_id_user filter has been used for filtering
1191 ##If theslice record is from iotlab
1192 #if fixed_slicerec_dict['peer_authority'] is None:
1193 #login = fixed_slicerec_dict['hrn'].split(".")[1].split("_")[0]
1194 #return login, fixed_slicerec_dict
1195 return fixed_slicerec_dict
1199 def GetSlices(self, slice_filter = None, slice_filter_type = None, \
1201 """ Get the slice records from the iotlab db and add lease information
1204 :param slice_filter: can be the slice hrn or slice record id in the db
1205 depending on the slice_filter_type.
1206 :param slice_filter_type: defines the type of the filtering used, Can be
1207 either 'slice_hrn' or "record_id'.
1208 :type slice_filter: string
1209 :type slice_filter_type: string
1210 :returns: a slice dict if slice_filter and slice_filter_type
1211 are specified and a matching entry is found in the db. The result
1212 is put into a list.Or a list of slice dictionnaries if no filters are
1218 authorized_filter_types_list = ['slice_hrn', 'record_id_user']
1219 return_slicerec_dictlist = []
1221 #First try to get information on the slice based on the filter provided
1222 if slice_filter_type in authorized_filter_types_list:
1223 fixed_slicerec_dict = \
1224 self._get_slice_records(slice_filter, slice_filter_type)
1225 slice_hrn = fixed_slicerec_dict['hrn']
1227 logger.debug(" IOTLABDRIVER \tGetSlices login %s \
1228 slice record %s slice_filter %s \
1229 slice_filter_type %s " %(login, \
1230 fixed_slicerec_dict, slice_filter, \
1234 #Now we have the slice record fixed_slicerec_dict, get the
1235 #jobs associated to this slice
1238 leases_list = self.GetLeases(login = login)
1239 #If no job is running or no job scheduled
1240 #return only the slice record
1241 if leases_list == [] and fixed_slicerec_dict:
1242 return_slicerec_dictlist.append(fixed_slicerec_dict)
1244 #If several jobs for one slice , put the slice record into
1245 # each lease information dict
1248 for lease in leases_list :
1250 logger.debug("IOTLABDRIVER.PY \tGetSlices slice_filter %s \
1251 \ lease['slice_hrn'] %s" \
1252 %(slice_filter, lease['slice_hrn']))
1253 if lease['slice_hrn'] == slice_hrn:
1254 slicerec_dict['slice_hrn'] = lease['slice_hrn']
1255 slicerec_dict['hrn'] = lease['slice_hrn']
1256 slicerec_dict['user'] = lease['user']
1257 slicerec_dict['oar_job_id'] = lease['lease_id']
1258 slicerec_dict.update({'list_node_ids':{'hostname':lease['reserved_nodes']}})
1259 slicerec_dict.update({'node_ids':lease['reserved_nodes']})
1261 #Update lease dict with the slice record
1262 if fixed_slicerec_dict:
1263 fixed_slicerec_dict['oar_job_id'] = []
1264 fixed_slicerec_dict['oar_job_id'].append(slicerec_dict['oar_job_id'])
1265 slicerec_dict.update(fixed_slicerec_dict)
1266 #slicerec_dict.update({'hrn':\
1267 #str(fixed_slicerec_dict['slice_hrn'])})
1269 return_slicerec_dictlist.append(slicerec_dict)
1270 logger.debug("IOTLABDRIVER.PY \tGetSlices \
1271 OHOHOHOH %s" %(return_slicerec_dictlist ))
1273 logger.debug("IOTLABDRIVER.PY \tGetSlices \
1274 slicerec_dict %s return_slicerec_dictlist %s \
1275 lease['reserved_nodes'] \
1276 %s" %(slicerec_dict, return_slicerec_dictlist, \
1277 lease['reserved_nodes'] ))
1279 logger.debug("IOTLABDRIVER.PY \tGetSlices RETURN \
1280 return_slicerec_dictlist %s" \
1281 %(return_slicerec_dictlist))
1283 return return_slicerec_dictlist
1287 #Get all slices from the iotlab sfa database ,
1288 #put them in dict format
1289 #query_slice_list = dbsession.query(RegRecord).all()
1290 query_slice_list = dbsession.query(RegSlice).options(joinedload('reg_researchers')).all()
1292 for record in query_slice_list:
1293 tmp = record.__dict__
1294 tmp['reg_researchers'] = tmp['reg_researchers'][0].__dict__
1295 #del tmp['reg_researchers']['_sa_instance_state']
1296 return_slicerec_dictlist.append(tmp)
1297 #return_slicerec_dictlist.append(record.__dict__)
1299 #Get all the jobs reserved nodes
1300 leases_list = self.GetReservedNodes()
1303 for fixed_slicerec_dict in return_slicerec_dictlist:
1305 #Check if the slice belongs to a iotlab user
1306 if fixed_slicerec_dict['peer_authority'] is None:
1307 owner = fixed_slicerec_dict['hrn'].split(".")[1].split("_")[0]
1310 for lease in leases_list:
1311 if owner == lease['user']:
1312 slicerec_dict['oar_job_id'] = lease['lease_id']
1314 #for reserved_node in lease['reserved_nodes']:
1315 logger.debug("IOTLABDRIVER.PY \tGetSlices lease %s "\
1318 slicerec_dict.update({'node_ids':lease['reserved_nodes']})
1319 slicerec_dict.update({'list_node_ids':{'hostname':lease['reserved_nodes']}})
1320 slicerec_dict.update(fixed_slicerec_dict)
1321 #slicerec_dict.update({'hrn':\
1322 #str(fixed_slicerec_dict['slice_hrn'])})
1323 #return_slicerec_dictlist.append(slicerec_dict)
1324 fixed_slicerec_dict.update(slicerec_dict)
1326 logger.debug("IOTLABDRIVER.PY \tGetSlices RETURN \
1327 return_slicerec_dictlist %s \slice_filter %s " \
1328 %(return_slicerec_dictlist, slice_filter))
1330 return return_slicerec_dictlist
1334 #Update slice unused, therefore sfa_fields_to_iotlab_fields unused
1337 #def sfa_fields_to_iotlab_fields(sfa_type, hrn, record):
1342 ##for field in record:
1343 ## iotlab_record[field] = record[field]
1345 #if sfa_type == "slice":
1346 ##instantion used in get_slivers ?
1347 #if not "instantiation" in iotlab_record:
1348 #iotlab_record["instantiation"] = "iotlab-instantiated"
1349 ##iotlab_record["hrn"] = hrn_to_pl_slicename(hrn)
1350 ##Unused hrn_to_pl_slicename because Iotlab's hrn already
1351 ##in the appropriate form SA 23/07/12
1352 #iotlab_record["hrn"] = hrn
1353 #logger.debug("IOTLABDRIVER.PY sfa_fields_to_iotlab_fields \
1354 #iotlab_record %s " %(iotlab_record['hrn']))
1355 #if "url" in record:
1356 #iotlab_record["url"] = record["url"]
1357 #if "description" in record:
1358 #iotlab_record["description"] = record["description"]
1359 #if "expires" in record:
1360 #iotlab_record["expires"] = int(record["expires"])
1362 ##nodes added by OAR only and then imported to SFA
1363 ##elif type == "node":
1364 ##if not "hostname" in iotlab_record:
1365 ##if not "hostname" in record:
1366 ##raise MissingSfaInfo("hostname")
1367 ##iotlab_record["hostname"] = record["hostname"]
1368 ##if not "model" in iotlab_record:
1369 ##iotlab_record["model"] = "geni"
1371 ##One authority only
1372 ##elif type == "authority":
1373 ##iotlab_record["login_base"] = hrn_to_iotlab_login_base(hrn)
1375 ##if not "name" in iotlab_record:
1376 ##iotlab_record["name"] = hrn
1378 ##if not "abbreviated_name" in iotlab_record:
1379 ##iotlab_record["abbreviated_name"] = hrn
1381 ##if not "enabled" in iotlab_record:
1382 ##iotlab_record["enabled"] = True
1384 ##if not "is_public" in iotlab_record:
1385 ##iotlab_record["is_public"] = True
1387 #return iotlab_record