4 from datetime import datetime
5 from dateutil import tz
6 from time import strftime,gmtime
8 from sfa.util.faults import MissingSfaInfo , SliverDoesNotExist
9 from sfa.util.sfalogging import logger
10 from sfa.util.defaultdict import defaultdict
12 from sfa.storage.record import Record
13 from sfa.storage.alchemy import dbsession
14 from sfa.storage.model import RegRecord
17 from sfa.trust.certificate import *
18 from sfa.trust.credential import *
19 from sfa.trust.gid import GID
21 from sfa.managers.driver import Driver
22 from sfa.rspecs.version_manager import VersionManager
23 from sfa.rspecs.rspec import RSpec
25 from sfa.util.xrn import hrn_to_urn, urn_to_sliver_id
26 from sfa.util.plxrn import slicename_to_hrn, hostname_to_hrn, hrn_to_pl_slicename
28 ## thierry: everything that is API-related (i.e. handling incoming requests)
30 # SlabDriver should be really only about talking to the senslab testbed
32 ## thierry : please avoid wildcard imports :)
33 from sfa.senslab.OARrestapi import OARrestapi
34 from sfa.senslab.LDAPapi import LDAPapi
36 from sfa.senslab.parsing import parse_filter
37 from sfa.senslab.slabpostgres import SlabDB, slab_dbsession,SliceSenslab
38 from sfa.senslab.slabaggregate import SlabAggregate
39 from sfa.senslab.slabslices import SlabSlices
41 def list_to_dict(recs, key):
43 convert a list of dictionaries into a dictionary keyed on the
44 specified dictionary key
47 keys = [rec[key] for rec in recs]
48 return dict(zip(keys, recs))
51 # this inheritance scheme is so that the driver object can receive
52 # GetNodes or GetSites sorts of calls directly
53 # and thus minimize the differences in the managers with the pl version
54 class SlabDriver(Driver):
56 def __init__(self, config):
57 Driver.__init__ (self, config)
59 self.hrn = config.SFA_INTERFACE_HRN
61 self.root_auth = config.SFA_REGISTRY_ROOT_AUTH
64 print >>sys.stderr, "\r\n_____________ SFA SENSLAB DRIVER \r\n"
66 self.oar = OARrestapi()
68 self.time_format = "%Y-%m-%d %H:%M:%S"
69 self.db = SlabDB(config)
73 def sliver_status(self,slice_urn,slice_hrn):
74 # receive a status request for slice named urn/hrn urn:publicid:IDN+senslab+nturro_slice hrn senslab.nturro_slice
75 # shall return a structure as described in
76 # http://groups.geni.net/geni/wiki/GAPI_AM_API_V2#SliverStatus
77 # NT : not sure if we should implement this or not, but used by sface.
80 sl = self.GetSlices(slice_filter= slice_hrn, filter_type = 'slice_hrn')
82 raise SliverDoesNotExist("%s slice_hrn" % (slice_hrn))
84 print >>sys.stderr, "\r\n \r\n_____________ Sliver status urn %s hrn %s sl %s \r\n " %(slice_urn,slice_hrn,sl)
85 if sl['oar_job_id'] is not -1:
87 # report about the local nodes only
88 nodes_all = self.GetNodes({'hostname':sl['node_ids']},
89 ['node_id', 'hostname','site','boot_state'])
90 nodeall_byhostname = dict([(n['hostname'], n) for n in nodes_all])
91 nodes = sl['node_ids']
93 raise SliverDoesNotExist("No slivers allocated ")
97 top_level_status = 'unknown'
99 top_level_status = 'ready'
100 result['geni_urn'] = slice_urn
101 result['pl_login'] = sl['job_user']
102 #result['slab_login'] = sl['job_user']
104 timestamp = float(sl['startTime']) + float(sl['walltime'])
105 result['pl_expires'] = strftime(self.time_format, gmtime(float(timestamp)))
106 #result['slab_expires'] = strftime(self.time_format, gmtime(float(timestamp)))
111 #res['slab_hostname'] = node['hostname']
112 #res['slab_boot_state'] = node['boot_state']
114 res['pl_hostname'] = nodeall_byhostname[node]['hostname']
115 res['pl_boot_state'] = nodeall_byhostname[node]['boot_state']
116 res['pl_last_contact'] = strftime(self.time_format, gmtime(float(timestamp)))
117 sliver_id = urn_to_sliver_id(slice_urn, sl['record_id_slice'],nodeall_byhostname[node]['node_id'] )
118 res['geni_urn'] = sliver_id
119 if nodeall_byhostname[node]['boot_state'] == 'Alive':
121 res['geni_status'] = 'ready'
123 res['geni_status'] = 'failed'
124 top_level_status = 'failed'
126 res['geni_error'] = ''
128 resources.append(res)
130 result['geni_status'] = top_level_status
131 result['geni_resources'] = resources
132 print >>sys.stderr, "\r\n \r\n_____________ Sliver status resources %s res %s \r\n " %(resources,res)
136 def create_sliver (self, slice_urn, slice_hrn, creds, rspec_string, users, options):
137 print>>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver "
138 aggregate = SlabAggregate(self)
140 slices = SlabSlices(self)
141 peer = slices.get_peer(slice_hrn)
142 sfa_peer = slices.get_sfa_peer(slice_hrn)
145 if not isinstance(creds, list):
149 slice_record = users[0].get('slice_record', {})
152 rspec = RSpec(rspec_string)
153 print>>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver ============================rspec.version %s " %(rspec.version)
156 # ensure site record exists?
157 # ensure slice record exists
158 slice = slices.verify_slice(slice_hrn, slice_record, peer, sfa_peer, options=options)
159 requested_attributes = rspec.version.get_slice_attributes()
161 if requested_attributes:
162 for attrib_dict in requested_attributes:
163 if 'timeslot' in attrib_dict and attrib_dict['timeslot'] is not None:
164 slice.update({'timeslot':attrib_dict['timeslot']})
165 print >>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver ..... slice %s " %(slice)
166 # ensure person records exists
167 persons = slices.verify_persons(slice_hrn, slice, users, peer, sfa_peer, options=options)
168 # ensure slice attributes exists?
171 # add/remove slice from nodes
172 print >>sys.stderr, "\r\n \r\n \t=============================== SLABDRIVER.PY create_sliver ..... "
174 requested_slivers = [node.get('component_name') for node in rspec.version.get_nodes_with_slivers()]
175 print >>sys.stderr, "\r\n \r\n \t=============================== ........... requested_slivers ============================requested_slivers %s " %(requested_slivers)
176 nodes = slices.verify_slice_nodes(slice, requested_slivers, peer)
179 return aggregate.get_rspec(slice_xrn=slice_urn, version=rspec.version)
182 def delete_sliver (self, slice_urn, slice_hrn, creds, options):
184 slice = self.GetSlices(slice_filter= slice_hrn, filter_type = 'slice_hrn')
185 print>>sys.stderr, "\r\n \r\n \t\t SLABDRIVER.PY delete_sliver slice %s" %(slice)
189 slices = SlabSlices(self)
190 # determine if this is a peer slice
191 # xxx I wonder if this would not need to use PlSlices.get_peer instead
192 # in which case plc.peers could be deprecated as this here
193 # is the only/last call to this last method in plc.peers
194 peer = slices.get_peer(slice_hrn)
197 self.UnBindObjectFromPeer('slice', slice['record_id_slice'], peer)
198 self.DeleteSliceFromNodes(slice)
201 self.BindObjectToPeer('slice', slice['slice_id'], peer, slice['peer_slice_id'])
205 def AddSlice(self, slice_record):
206 slab_slice = SliceSenslab( slice_hrn = slice_record['slice_hrn'], record_id_slice= slice_record['record_id_slice'] , record_id_user= slice_record['record_id_user'], peer_authority = slice_record['peer_authority'])
207 print>>sys.stderr, "\r\n \r\n \t\t\t =======SLABDRIVER.PY AddSlice slice_record %s slab_slice %s" %(slice_record,slab_slice)
208 slab_dbsession.add(slab_slice)
209 slab_dbsession.commit()
212 # first 2 args are None in case of resource discovery
213 def list_resources (self, slice_urn, slice_hrn, creds, options):
214 #cached_requested = options.get('cached', True)
216 version_manager = VersionManager()
217 # get the rspec's return format from options
218 rspec_version = version_manager.get_version(options.get('geni_rspec_version'))
219 version_string = "rspec_%s" % (rspec_version)
221 #panos adding the info option to the caching key (can be improved)
222 if options.get('info'):
223 version_string = version_string + "_"+options.get('info', 'default')
225 # look in cache first
226 #if cached_requested and self.cache and not slice_hrn:
227 #rspec = self.cache.get(version_string)
229 #logger.debug("SlabDriver.ListResources: returning cached advertisement")
232 #panos: passing user-defined options
234 aggregate = SlabAggregate(self)
235 origin_hrn = Credential(string=creds[0]).get_gid_caller().get_hrn()
236 options.update({'origin_hrn':origin_hrn})
237 rspec = aggregate.get_rspec(slice_xrn=slice_urn, version=rspec_version,
239 print>>sys.stderr, " \r\n \r\n \t SLABDRIVER list_resources rspec "
241 #if self.cache and not slice_hrn:
242 #logger.debug("Slab.ListResources: stores advertisement in cache")
243 #self.cache.add(version_string, rspec)
248 def list_slices (self, creds, options):
249 # look in cache first
251 #slices = self.cache.get('slices')
253 #logger.debug("PlDriver.list_slices returns from cache")
257 print>>sys.stderr, " \r\n \t\t SLABDRIVER.PY list_slices"
258 slices = self.GetSlices()
259 slice_hrns = [slicename_to_hrn(self.hrn, slice['slice_hrn']) for slice in slices]
260 slice_urns = [hrn_to_urn(slice_hrn, 'slice') for slice_hrn in slice_hrns]
264 #logger.debug ("SlabDriver.list_slices stores value in cache")
265 #self.cache.add('slices', slice_urns)
269 #No site or node register supported
270 def register (self, sfa_record, hrn, pub_key):
271 type = sfa_record['type']
272 slab_record = self.sfa_fields_to_slab_fields(type, hrn, sfa_record)
276 acceptable_fields=['url', 'instantiation', 'name', 'description']
277 for key in slab_record.keys():
278 if key not in acceptable_fields:
280 print>>sys.stderr, " \r\n \t\t SLABDRIVER.PY register"
281 slices = self.GetSlices(slice_filter =slab_record['hrn'], filter_type = 'slice_hrn')
283 pointer = self.AddSlice(slab_record)
285 pointer = slices[0]['slice_id']
288 persons = self.GetPersons([sfa_record['hrn']])
290 pointer = self.AddPerson(dict(sfa_record))
293 pointer = persons[0]['person_id']
295 #Does this make sense to senslab ?
296 #if 'enabled' in sfa_record and sfa_record['enabled']:
297 #self.UpdatePerson(pointer, {'enabled': sfa_record['enabled']})
299 # add this person to the site only if she is being added for the first
300 # time by sfa and doesont already exist in plc
301 if not persons or not persons[0]['site_ids']:
302 login_base = get_leaf(sfa_record['authority'])
303 self.AddPersonToSite(pointer, login_base)
305 # What roles should this user have?
306 self.AddRoleToPerson('user', pointer)
309 self.AddPersonKey(pointer, {'key_type' : 'ssh', 'key' : pub_key})
311 #No node adding outside OAR
315 #No site or node record update allowed
316 def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
317 pointer = old_sfa_record['pointer']
318 type = old_sfa_record['type']
320 # new_key implemented for users only
321 if new_key and type not in [ 'user' ]:
322 raise UnknownSfaType(type)
324 #if (type == "authority"):
325 #self.shell.UpdateSite(pointer, new_sfa_record)
328 slab_record=self.sfa_fields_to_slab_fields(type, hrn, new_sfa_record)
329 if 'name' in slab_record:
330 slab_record.pop('name')
331 self.UpdateSlice(pointer, slab_record)
335 all_fields = new_sfa_record
336 for key in all_fields.keys():
337 if key in ['first_name', 'last_name', 'title', 'email',
338 'password', 'phone', 'url', 'bio', 'accepted_aup',
340 update_fields[key] = all_fields[key]
341 self.UpdatePerson(pointer, update_fields)
344 # must check this key against the previous one if it exists
345 persons = self.GetPersons([pointer], ['key_ids'])
347 keys = person['key_ids']
348 keys = self.GetKeys(person['key_ids'])
350 # Delete all stale keys
353 if new_key != key['key']:
354 self.DeleteKey(key['key_id'])
358 self.AddPersonKey(pointer, {'key_type': 'ssh', 'key': new_key})
364 def remove (self, sfa_record):
365 type=sfa_record['type']
366 hrn=sfa_record['hrn']
367 record_id= sfa_record['record_id']
369 username = hrn.split(".")[len(hrn.split(".")) -1]
371 persons = self.GetPersons(username)
372 # only delete this person if he has site ids. if he doesnt, it probably means
373 # he was just removed from a site, not actually deleted
374 if persons and persons[0]['site_ids']:
375 self.DeletePerson(username)
376 elif type == 'slice':
377 if self.GetSlices(slice_filter = hrn, filter_type = 'slice_hrn'):
378 self.DeleteSlice(hrn)
380 #elif type == 'authority':
381 #if self.GetSites(pointer):
382 #self.DeleteSite(pointer)
386 def GetPeers (self,auth = None, peer_filter=None, return_fields=None):
388 existing_records = {}
389 existing_hrns_by_types= {}
390 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers auth = %s, peer_filter %s, return_field %s " %(auth , peer_filter, return_fields)
391 all_records = dbsession.query(RegRecord).filter(RegRecord.type.like('%authority%')).all()
392 for record in all_records:
393 existing_records[(record.hrn,record.type)] = record
394 if record.type not in existing_hrns_by_types:
395 existing_hrns_by_types[record.type] = [record.hrn]
396 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers \t NOT IN existing_hrns_by_types %s " %( existing_hrns_by_types)
399 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers \t INNN type %s hrn %s " %( record.type,record.hrn )
400 existing_hrns_by_types[record.type].append(record.hrn)
401 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers \t INNN existing_hrns_by_types %s " %( existing_hrns_by_types)
402 #existing_hrns_by_types.update({record.type:(existing_hrns_by_types[record.type].append(record.hrn))})
404 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers existing_hrns_by_types %s " %( existing_hrns_by_types)
408 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers existing_hrns_by_types['authority+sa'] %s \t\t existing_records %s " %(existing_hrns_by_types['authority'],existing_records)
410 records_list.append(existing_records[(peer_filter,'authority')])
412 for hrn in existing_hrns_by_types['authority']:
413 records_list.append(existing_records[(hrn,'authority')])
415 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers records_list %s " %(records_list)
420 return_records = records_list
421 if not peer_filter and not return_fields:
423 #return_records = parse_filter(records_list,peer_filter, 'peers', return_fields)
425 print >>sys.stderr, "\r\n \r\n SLABDRIVER GetPeers return_records %s " %(return_records)
426 return return_records
429 #TODO : Handling OR request in parse_records instead of the for loop
430 #over the records' list
431 def GetPersons(self, person_filter=None, return_fields=None):
433 person_filter should be a list of dictionnaries when not set to None.
434 Returns a list of users found.
437 print>>sys.stderr, "\r\n \r\n \t\t\t GetPersons person_filter %s" %(person_filter)
439 if person_filter and isinstance(person_filter,list):
440 #If we are looking for a list of users (list of dict records)
441 #Usually the list contains only one user record
442 for f in person_filter:
443 person = self.ldap.ldapSearch(f)
444 person_list.append(person)
447 person_list = self.ldap.ldapSearch()
450 #person_list = self.ldap.ldapFindHrn({'authority': self.root_auth })
452 ##if person_filter and isinstance(person_filter, dict):
453 ##for k in person_filter.keys():
454 ##if k in person_list[0].keys():
457 #return_person_list = parse_filter(person_list,person_filter ,'persons', return_fields)
458 #if return_person_list:
459 #print>>sys.stderr, " \r\n GetPersons person_filter %s return_fields %s " %(person_filter,return_fields)
460 #return return_person_list
462 def GetTimezone(self):
463 server_timestamp,server_tz = self.oar.parser.SendRequest("GET_timezone")
464 return server_timestamp,server_tz
467 def DeleteJobs(self, job_id, slice_hrn):
470 username = slice_hrn.split(".")[-1].rstrip("_slice")
472 reqdict['method'] = "delete"
473 reqdict['strval'] = str(job_id)
474 answer = self.oar.POSTRequestToOARRestAPI('DELETE_jobs_id',reqdict,username)
475 print>>sys.stderr, "\r\n \r\n jobid DeleteJobs %s " %(answer)
478 def GetJobs(self,job_id= None, resources=True,return_fields=None, username = None):
479 #job_resources=['reserved_resources', 'assigned_resources','job_id', 'job_uri', 'assigned_nodes',\
481 #assigned_res = ['resource_id', 'resource_uri']
482 #assigned_n = ['node', 'node_uri']
484 if job_id and resources is False:
486 node_list_k = 'assigned_network_address'
488 if job_id and resources :
489 req = "GET_jobs_id_resources"
490 node_list_k = 'reserved_resources'
492 #Get job info from OAR
493 job_info = self.oar.parser.SendRequest(req, job_id, username)
494 print>>sys.stderr, "\r\n \r\n \t\t GetJobs %s " %(job_info)
496 if 'state' in job_info :
497 if job_info['state'] == 'Terminated':
498 print>>sys.stderr, "\r\n \r\n \t\t GetJobs TERMINELEBOUSIN "
500 if job_info['state'] == 'Error':
501 print>>sys.stderr, "\r\n \r\n \t\t GetJobs ERROR message %s " %(job_info)
504 #Get a dict of nodes . Key :hostname of the node
505 node_list = self.GetNodes()
506 node_hostname_list = []
507 for node in node_list:
508 node_hostname_list.append(node['hostname'])
509 node_dict = dict(zip(node_hostname_list,node_list))
511 liste =job_info[node_list_k]
512 for k in range(len(liste)):
513 job_info[node_list_k][k] = node_dict[job_info[node_list_k][k]]['hostname']
515 #Replaces the previous entry "assigned_network_address" / "reserved_resources"
517 job_info.update({'node_ids':job_info[node_list_k]})
518 del job_info[node_list_k]
522 print>>sys.stderr, "\r\n \r\n \t\t GetJobs KEYERROR "
524 def GetReservedNodes(self):
525 # this function returns a list of all the nodes already involved in an oar job
527 jobs=self.oar.parser.SendRequest("GET_jobs_details")
530 nodes=j['assigned_network_address']+nodes
533 def GetNodes(self,node_filter= None, return_fields=None):
534 node_dict =self.oar.parser.SendRequest("GET_resources_full")
536 return_node_list = []
537 if not (node_filter or return_fields):
538 return_node_list = node_dict.values()
539 return return_node_list
541 return_node_list= parse_filter(node_dict.values(),node_filter ,'node', return_fields)
542 return return_node_list
545 def GetSites(self, site_filter = None, return_fields=None):
546 site_dict =self.oar.parser.SendRequest("GET_sites")
547 return_site_list = []
548 if not ( site_filter or return_fields):
549 return_site_list = site_dict.values()
550 return return_site_list
552 return_site_list = parse_filter(site_dict.values(), site_filter,'site', return_fields)
553 return return_site_list
556 def GetSlices(self,slice_filter = None, filter_type = None, return_fields=None):
557 return_slice_list = []
560 ftypes = ['slice_hrn', 'record_id_user']
561 if filter_type and filter_type in ftypes:
562 if filter_type == 'slice_hrn':
563 slicerec = slab_dbsession.query(SliceSenslab).filter_by(slice_hrn = slice_filter).first()
564 if filter_type == 'record_id_user':
565 slicerec = slab_dbsession.query(SliceSenslab).filter_by(record_id_user = slice_filter).first()
568 rec = slicerec.dumpquerytodict()
569 login = slicerec.slice_hrn.split(".")[1].split("_")[0]
570 #print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY slicerec GetSlices %s " %(slicerec)
571 if slicerec.oar_job_id is not -1:
572 rslt = self.GetJobs( slicerec.oar_job_id, resources=False, username = login )
573 #print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY GetSlices GetJobs %s " %(rslt)
576 rec.update({'hrn':str(rec['slice_hrn'])})
577 #If GetJobs is empty, this means the job is now in the 'Terminated' state
578 #Update the slice record
580 self.db.update_job(slice_filter, job_id = -1)
581 rec['oar_job_id'] = -1
582 rec.update({'hrn':str(rec['slice_hrn'])})
585 rec['node_ids'] = rec['node_list']
589 #print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY GetSlices rec %s" %(rec)
595 return_slice_list = slab_dbsession.query(SliceSenslab).all()
597 print >>sys.stderr, " \r\n \r\n \tSLABDRIVER.PY GetSlices slices %s slice_filter %s " %(return_slice_list,slice_filter)
600 #return_slice_list = parse_filter(sliceslist, slice_filter,'slice', return_fields)
604 return return_slice_list
609 def testbed_name (self): return "senslab2"
611 # 'geni_request_rspec_versions' and 'geni_ad_rspec_versions' are mandatory
612 def aggregate_version (self):
613 version_manager = VersionManager()
614 ad_rspec_versions = []
615 request_rspec_versions = []
616 for rspec_version in version_manager.versions:
617 if rspec_version.content_type in ['*', 'ad']:
618 ad_rspec_versions.append(rspec_version.to_dict())
619 if rspec_version.content_type in ['*', 'request']:
620 request_rspec_versions.append(rspec_version.to_dict())
622 'testbed':self.testbed_name(),
623 'geni_request_rspec_versions': request_rspec_versions,
624 'geni_ad_rspec_versions': ad_rspec_versions,
633 # Convert SFA fields to PLC fields for use when registering up updating
634 # registry record in the PLC database
636 # @param type type of record (user, slice, ...)
637 # @param hrn human readable name
638 # @param sfa_fields dictionary of SFA fields
639 # @param slab_fields dictionary of PLC fields (output)
641 def sfa_fields_to_slab_fields(self, type, hrn, record):
643 def convert_ints(tmpdict, int_fields):
644 for field in int_fields:
646 tmpdict[field] = int(tmpdict[field])
649 #for field in record:
650 # slab_record[field] = record[field]
653 #instantion used in get_slivers ?
654 if not "instantiation" in slab_record:
655 slab_record["instantiation"] = "senslab-instantiated"
656 slab_record["hrn"] = hrn_to_pl_slicename(hrn)
657 print >>sys.stderr, "\r\n \r\n \t SLABDRIVER.PY sfa_fields_to_slab_fields slab_record %s hrn_to_pl_slicename(hrn) hrn %s " %(slab_record['hrn'], hrn)
659 slab_record["url"] = record["url"]
660 if "description" in record:
661 slab_record["description"] = record["description"]
662 if "expires" in record:
663 slab_record["expires"] = int(record["expires"])
665 #nodes added by OAR only and then imported to SFA
666 #elif type == "node":
667 #if not "hostname" in slab_record:
668 #if not "hostname" in record:
669 #raise MissingSfaInfo("hostname")
670 #slab_record["hostname"] = record["hostname"]
671 #if not "model" in slab_record:
672 #slab_record["model"] = "geni"
675 #elif type == "authority":
676 #slab_record["login_base"] = hrn_to_slab_login_base(hrn)
678 #if not "name" in slab_record:
679 #slab_record["name"] = hrn
681 #if not "abbreviated_name" in slab_record:
682 #slab_record["abbreviated_name"] = hrn
684 #if not "enabled" in slab_record:
685 #slab_record["enabled"] = True
687 #if not "is_public" in slab_record:
688 #slab_record["is_public"] = True
693 def LaunchExperimentOnOAR(self, slice_dict, added_nodes, slice_user=None):
699 slice_name = slice_dict['name']
701 slot = slice_dict['timeslot']
702 print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR slot %s " %(slot)
704 #Running on default parameters
705 #XP immediate , 10 mins
706 slot = {'date':None,'start_time':None, 'timezone':None,'duration':None }#10 min
709 reqdict['property'] ="network_address in ("
710 for node in added_nodes:
711 #Get the ID of the node : remove the root auth and put the site in a separate list
713 # NT: it's not clear for me if the nodenames will have the senslab prefix
714 # so lets take the last part only, for now.
716 #if s[0] == self.root_auth :
717 # Again here it's not clear if nodes will be prefixed with <site>_, lets split and tanke the last part for now.
718 s=lastpart.split("_")
720 reqdict['property'] += "'"+ nodeid +"', "
721 nodeid_list.append(nodeid)
722 #site_list.append( l[0] )
725 reqdict['property'] = reqdict['property'][0: len( reqdict['property'])-2] +")"
726 reqdict['resource'] ="network_address="+ str(len(nodeid_list))
729 walltime = slot['duration'].split(":")
730 # Fixing the walltime by adding a few delays. First put the walltime in seconds
731 # oarAdditionalDelay = 20; additional delay for /bin/sleep command to
732 # take in account prologue and epilogue scripts execution
733 # int walltimeAdditionalDelay = 120; additional delay
735 desired_walltime = int(walltime[0])*3600 + int(walltime[1]) * 60 + int(walltime[2])
736 total_walltime = desired_walltime + 140 #+2 min 20
737 sleep_walltime = desired_walltime + 20 #+20 sec
738 print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR desired_walltime %s total_walltime %s sleep_walltime %s " %(desired_walltime,total_walltime,sleep_walltime)
739 #Put the walltime back in str form
741 walltime[0] = str(total_walltime / 3600)
742 total_walltime = total_walltime - 3600 * int(walltime[0])
743 #Get the remaining minutes
744 walltime[1] = str(total_walltime / 60)
745 total_walltime = total_walltime - 60 * int(walltime[1])
747 walltime[2] = str(total_walltime)
748 print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR walltime %s " %(walltime)
750 reqdict['resource']+= ",walltime=" + str(walltime[0]) + ":" + str(walltime[1]) + ":" + str(walltime[2])
751 reqdict['script_path'] = "/bin/sleep " + str(sleep_walltime)
753 reqdict['resource']+= ",walltime=" + str(00) + ":" + str(12) + ":" + str(20) #+2 min 20
754 reqdict['script_path'] = "/bin/sleep 620" #+20 sec
755 #In case of a scheduled experiment (not immediate)
756 #To run an XP immediately, don't specify date and time in RSpec
757 #They will be set to None.
758 if slot['date'] and slot['start_time']:
759 if slot['timezone'] is '' or slot['timezone'] is None:
760 #assume it is server timezone
761 server_timestamp,server_tz = self.GetTimezone()
762 from_zone=tz.gettz(server_tz)
763 print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR timezone not specified server_tz %s from_zone %s" %(server_tz,from_zone)
765 #Get zone of the user from the reservation time given in the rspec
766 from_zone = tz.gettz(slot['timezone'])
768 date = str(slot['date']) + " " + str(slot['start_time'])
769 user_datetime = datetime.datetime.strptime(date, self.time_format)
770 user_datetime = user_datetime.replace(tzinfo = from_zone)
774 utc_date = user_datetime.astimezone(to_zone)
775 #Readable time accpeted by OAR
776 reqdict['reservation']= utc_date.strftime(self.time_format)
778 print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR reqdict['reservation'] %s " %(reqdict['reservation'])
782 # reservations are performed in the oar server timebase, so :
783 # 1- we get the server time(in UTC tz )/server timezone
784 # 2- convert the server UTC time in its timezone
785 # 3- add a custom delay to this time
786 # 4- convert this time to a readable form and it for the reservation request.
787 server_timestamp,server_tz = self.GetTimezone()
788 s_tz=tz.gettz(server_tz)
789 UTC_zone = tz.gettz("UTC")
790 #weird... datetime.fromtimestamp should work since we do from datetime import datetime
791 utc_server= datetime.datetime.fromtimestamp(float(server_timestamp)+20,UTC_zone)
792 server_localtime=utc_server.astimezone(s_tz)
794 print>>sys.stderr, "\r\n \r\n \t\tLaunchExperimentOnOAR server_timestamp %s server_tz %s slice_name %s added_nodes %s username %s reqdict %s " %(server_timestamp,server_tz,slice_name,added_nodes,slice_user, reqdict )
795 readable_time = server_localtime.strftime(self.time_format)
797 print >>sys.stderr," \r\n \r\n \t\t\t\tAPRES ParseTimezone readable_time %s timestanp %s " %(readable_time ,server_timestamp)
798 reqdict['reservation'] = readable_time
801 reqdict['type'] = "deploy"
802 reqdict['directory']= ""
803 reqdict['name']= "TestSandrine"
806 # first step : start the OAR job and update the job
807 print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR reqdict %s \r\n site_list %s" %(reqdict,site_list)
809 answer = self.oar.POSTRequestToOARRestAPI('POST_job',reqdict,slice_user)
810 print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR jobid %s " %(answer)
814 print>>sys.stderr, "\r\n AddSliceTonode Impossible to create job %s " %( answer)
817 print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR jobid %s added_nodes %s slice_user %s" %(jobid,added_nodes,slice_user)
818 self.db.update_job( slice_name, jobid ,added_nodes)
821 # second step : configure the experiment
822 # we need to store the nodes in a yaml (well...) file like this :
823 # [1,56,23,14,45,75] with name /tmp/sfa<jobid>.json
824 f=open('/tmp/sfa/'+str(jobid)+'.json','w')
826 f.write(str(added_nodes[0].strip('node')))
827 for node in added_nodes[1:len(added_nodes)] :
828 f.write(','+node.strip('node'))
832 # third step : call the senslab-experiment wrapper
833 #command= "java -jar target/sfa-1.0-jar-with-dependencies.jar "+str(jobid)+" "+slice_user
834 javacmdline="/usr/bin/java"
835 jarname="/opt/senslabexperimentwrapper/sfa-1.0-jar-with-dependencies.jar"
836 #ret=subprocess.check_output(["/usr/bin/java", "-jar", ", str(jobid), slice_user])
837 output = subprocess.Popen([javacmdline, "-jar", jarname, str(jobid), slice_user],stdout=subprocess.PIPE).communicate()[0]
839 print>>sys.stderr, "\r\n \r\n LaunchExperimentOnOAR wrapper returns %s " %(output)
843 #Delete the jobs and updates the job id in the senslab table
845 #Does not clear the node list
846 def DeleteSliceFromNodes(self, slice_record):
847 # Get user information
849 self.DeleteJobs(slice_record['oar_job_id'], slice_record['hrn'])
850 self.db.update_job(slice_record['hrn'], job_id = -1)
857 def augment_records_with_testbed_info (self, sfa_records):
858 return self.fill_record_info (sfa_records)
860 def fill_record_info(self, records):
862 Given a SFA record, fill in the senslab specific and SFA specific
863 fields in the record.
866 print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info 000000000 fill_record_info %s " %(records)
867 if not isinstance(records, list):
872 for record in parkour:
874 if str(record['type']) == 'slice':
875 #print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info \t \t record %s" %(record)
876 #sfatable = SfaTable()
878 #existing_records_by_id = {}
879 #all_records = dbsession.query(RegRecord).all()
880 #for rec in all_records:
881 #existing_records_by_id[rec.record_id] = rec
882 #print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info \t\t existing_records_by_id %s" %(existing_records_by_id[record['record_id']])
884 #recslice = self.db.find('slice',{'slice_hrn':str(record['hrn'])})
885 #recslice = slab_dbsession.query(SliceSenslab).filter_by(slice_hrn = str(record['hrn'])).first()
886 recslice = self.GetSlices(slice_filter = str(record['hrn']), filter_type = 'slice_hrn')
887 #print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info \t\t HOY HOY reclise %s" %(recslice)
888 #if isinstance(recslice,list) and len(recslice) == 1:
889 #recslice = recslice[0]
891 recuser = dbsession.query(RegRecord).filter_by(record_id = recslice['record_id_user']).first()
892 #existing_records_by_id[recslice['record_id_user']]
893 #print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info \t\t recuser %s" %(recuser)
896 record.update({'PI':[recuser.hrn],
897 'researcher': [recuser.hrn],
898 'name':record['hrn'],
899 'oar_job_id':recslice['oar_job_id'],
901 'person_ids':[recslice['record_id_user']],
902 'geni_urn':'', #For client_helper.py compatibility
903 'keys':'', #For client_helper.py compatibility
904 'key_ids':''}) #For client_helper.py compatibility
906 elif str(record['type']) == 'user':
907 #Add the data about slice
908 rec = self.GetSlices(slice_filter = record['record_id'], filter_type = 'record_id_user')
909 print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info USEEEEEEEEEERDESU! rec %s \r\n \t rec['record_id_user'] %s " %(rec,rec['record_id_user'])
910 #Append record in records list, therfore fetches user and slice info again(one more loop)
911 #Will update PIs and researcher for the slice
912 recuser = dbsession.query(RegRecord).filter_by(record_id = rec['record_id_user']).first()
913 rec.update({'PI':[recuser.hrn],
914 'researcher': [recuser.hrn],
915 'name':record['hrn'],
916 'oar_job_id':rec['oar_job_id'],
918 'person_ids':[rec['record_id_user']]})
919 #retourne une liste 100512
921 #GetPersons takes [] as filters
922 user_slab = self.GetPersons([{'hrn':recuser.hrn}])
925 rec.update({'type':'slice','hrn':rec['slice_hrn']})
926 record.update(user_slab[0])
927 #For client_helper.py compatibility
928 record.update( { 'geni_urn':'',
933 print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info ADDING SLICEINFO TO USER records %s" %(records)
935 print >>sys.stderr, "\r\n \t\t SLABDRIVER.PY fill_record_info OKrecords %s" %(records)
937 print >>sys.stderr, "\r\n \t\t SLABDRIVER fill_record_info EXCEPTION RECORDS : %s" %(records)
940 #self.fill_record_slab_info(records)
941 ##print >>sys.stderr, "\r\n \t\t after fill_record_slab_info %s" %(records)
942 #self.fill_record_sfa_info(records)
943 #print >>sys.stderr, "\r\n \t\t after fill_record_sfa_info"
949 #def update_membership_list(self, oldRecord, record, listName, addFunc, delFunc):
950 ## get a list of the HRNs tht are members of the old and new records
952 #oldList = oldRecord.get(listName, [])
955 #newList = record.get(listName, [])
957 ## if the lists are the same, then we don't have to update anything
958 #if (oldList == newList):
961 ## build a list of the new person ids, by looking up each person to get
965 #records = table.find({'type': 'user', 'hrn': newList})
967 #newIdList.append(rec['pointer'])
969 ## build a list of the old person ids from the person_ids field
971 #oldIdList = oldRecord.get("person_ids", [])
972 #containerId = oldRecord.get_pointer()
974 ## if oldRecord==None, then we are doing a Register, instead of an
977 #containerId = record.get_pointer()
979 ## add people who are in the new list, but not the oldList
980 #for personId in newIdList:
981 #if not (personId in oldIdList):
982 #addFunc(self.plauth, personId, containerId)
984 ## remove people who are in the old list, but not the new list
985 #for personId in oldIdList:
986 #if not (personId in newIdList):
987 #delFunc(self.plauth, personId, containerId)
989 #def update_membership(self, oldRecord, record):
990 #print >>sys.stderr, " \r\n \r\n ***SLABDRIVER.PY update_membership record ", record
991 #if record.type == "slice":
992 #self.update_membership_list(oldRecord, record, 'researcher',
993 #self.users.AddPersonToSlice,
994 #self.users.DeletePersonFromSlice)
995 #elif record.type == "authority":
1000 # I don't think you plan on running a component manager at this point
1001 # let me clean up the mess of ComponentAPI that is deprecated anyways