2 from httplib import HTTPConnection, HTTPException
5 #from time import gmtime, strftime
10 from sfa.util.config import Config
11 #from sfa.util.xrn import hrn_to_urn, get_authority, Xrn, get_leaf
13 from sfa.util.sfalogging import logger
16 OAR_REQUEST_POST_URI_DICT = {'POST_job':{'uri': '/oarapi/jobs.json'},
17 'DELETE_jobs_id':{'uri':'/oarapi/jobs/id.json'},
20 POST_FORMAT = {'json' : {'content':"application/json", 'object':json},}
22 #OARpostdatareqfields = {'resource' :"/nodes=", 'command':"sleep", \
23 #'workdir':"/home/", 'walltime':""}
30 def __init__(self, config_file = '/etc/sfa/oar_config.py'):
34 self.oarserver['uri'] = None
35 self.oarserver['postformat'] = 'json'
38 execfile(config_file, self.__dict__)
40 self.config_file = config_file
41 # path to configuration data
42 self.config_path = os.path.dirname(config_file)
45 raise IOError, "Could not find or load the configuration file: %s" \
47 #logger.setLevelDebug()
48 self.oarserver['ip'] = self.OAR_IP
49 self.oarserver['port'] = self.OAR_PORT
50 self.jobstates = ['Terminated', 'Hold', 'Waiting', 'toLaunch', \
51 'toError', 'toAckReservation', 'Launching', \
52 'Finishing', 'Running', 'Suspended', 'Resuming',\
55 self.parser = OARGETParser(self)
58 def GETRequestToOARRestAPI(self, request, strval=None ,next_page=None, username = None ):
59 self.oarserver['uri'] = \
60 OARGETParser.OARrequests_uri_dict[request]['uri']
61 #Get job details with username
62 if 'owner' in OARGETParser.OARrequests_uri_dict[request] and username:
63 self.oarserver['uri'] += OARGETParser.OARrequests_uri_dict[request]['owner'] + username
66 logger.debug("OARrestapi \tGETRequestToOARRestAPI %s" %(request))
68 self.oarserver['uri'] = self.oarserver['uri'].\
69 replace("id",str(strval))
72 self.oarserver['uri'] += next_page
75 headers['X-REMOTE_IDENT'] = username
77 print>>sys.stderr, " \r\n \t OARrestapi \tGETRequestToOARRestAPI %s" %( self.oarserver['uri'])
78 logger.debug("OARrestapi: \t GETRequestToOARRestAPI \
79 self.oarserver['uri'] %s strval %s" \
80 %(self.oarserver['uri'], strval))
82 #seems that it does not work if we don't add this
83 headers['content-length'] = '0'
85 conn = HTTPConnection(self.oarserver['ip'], \
86 self.oarserver['port'])
87 conn.request("GET", self.oarserver['uri'], data, headers)
88 resp = ( conn.getresponse()).read()
90 except HTTPException, error :
91 logger.log_exc("GET_OAR_SRVR : Problem with OAR server : %s " \
93 #raise ServerError("GET_OAR_SRVR : Could not reach OARserver")
95 js_dict = json.loads(resp)
96 #print "\r\n \t\t\t js_dict keys" , js_dict.keys(), " \r\n", js_dict
99 except ValueError, error:
100 logger.log_exc("Failed to parse Server Response: %s ERROR %s"\
102 #raise ServerError("Failed to parse Server Response:" + js)
105 def POSTRequestToOARRestAPI(self, request, datadict, username=None):
106 """ Used to post a job on OAR , along with data associated
111 #first check that all params for are OK
113 self.oarserver['uri'] = OAR_REQUEST_POST_URI_DICT[request]['uri']
116 logger.log_exc("OARrestapi \tPOSTRequestToOARRestAPI request not \
119 if datadict and 'strval' in datadict:
120 self.oarserver['uri'] = self.oarserver['uri'].replace("id", \
121 str(datadict['strval']))
122 del datadict['strval']
124 data = json.dumps(datadict)
125 headers = {'X-REMOTE_IDENT':username, \
126 'content-type': POST_FORMAT['json']['content'], \
127 'content-length':str(len(data))}
130 conn = HTTPConnection(self.oarserver['ip'], \
131 self.oarserver['port'])
132 conn.request("POST", self.oarserver['uri'], data, headers)
133 resp = (conn.getresponse()).read()
136 logger.log_exc("POSTRequestToOARRestAPI NotConnected ERROR: \
137 data %s \r\n \t\n \t\t headers %s uri %s" \
138 %(data,headers,self.oarserver['uri']))
140 #raise ServerError("POST_OAR_SRVR : error")
143 answer = json.loads(resp)
144 logger.debug("POSTRequestToOARRestAPI : answer %s" %(answer))
147 except ValueError, error:
148 logger.log_exc("Failed to parse Server Response: error %s \
150 #raise ServerError("Failed to parse Server Response:" + answer)
154 def AddOarNodeId(tuplelist, value):
155 """ Adds Oar internal node id to the nodes attributes """
157 tuplelist.append(('oar_id', int(value)))
160 def AddNodeNetworkAddr(dictnode, value):
161 #Inserts new key. The value associated is a tuple list
164 dictnode[node_id] = [('node_id', node_id),('hostname', node_id) ]
168 def AddNodeSite(tuplelist, value):
169 tuplelist.append(('site', str(value)))
171 def AddNodeRadio(tuplelist, value):
172 tuplelist.append(('radio', str(value)))
175 def AddMobility(tuplelist, value):
177 tuplelist.append(('mobile', 'False'))
179 tuplelist.append(('mobile', 'True'))
181 def AddPosX(tuplelist, value):
182 tuplelist.append(('posx', value))
184 def AddPosY(tuplelist, value):
185 tuplelist.append(('posy', value))
187 def AddPosZ(tuplelist, value):
188 tuplelist.append(('posz', value))
190 def AddBootState(tuplelist, value):
191 tuplelist.append(('boot_state', str(value)))
193 #Insert a new node into the dictnode dictionary
194 def AddNodeId(dictnode, value):
195 #Inserts new key. The value associated is a tuple list
198 dictnode[node_id] = [('node_id', node_id)]
201 def AddHardwareType(tuplelist, value):
202 value_list = value.split(':')
203 tuplelist.append(('archi', value_list[0]))
204 tuplelist.append(('radio', value_list[1]))
208 resources_fulljson_dict = {
209 'network_address' : AddNodeNetworkAddr,
211 'radio': AddNodeRadio,
212 'mobile': AddMobility,
216 'archi':AddHardwareType,
217 'state':AddBootState,
222 def __init__(self, srv) :
223 self.version_json_dict = {
224 'api_version' : None , 'apilib_version' :None,\
225 'api_timezone': None, 'api_timestamp': None, 'oar_version': None ,}
226 self.config = Config()
227 self.interface_hrn = self.config.SFA_INTERFACE_HRN
228 self.timezone_json_dict = {
229 'timezone': None, 'api_timestamp': None, }
230 self.jobs_json_dict = {
231 'total' : None, 'links' : [],\
232 'offset':None , 'items' : [], }
233 self.jobs_table_json_dict = self.jobs_json_dict
234 self.jobs_details_json_dict = self.jobs_json_dict
236 self.node_dictlist = {}
239 self.SendRequest("GET_version")
245 def ParseVersion(self) :
247 #print >>sys.stderr, self.raw_json
248 if 'oar_version' in self.raw_json :
249 self.version_json_dict.update(api_version = \
250 self.raw_json['api_version'],
251 apilib_version = self.raw_json['apilib_version'],
252 api_timezone = self.raw_json['api_timezone'],
253 api_timestamp = self.raw_json['api_timestamp'],
254 oar_version = self.raw_json['oar_version'] )
256 self.version_json_dict.update(api_version = self.raw_json['api'] ,
257 apilib_version = self.raw_json['apilib'],
258 api_timezone = self.raw_json['api_timezone'],
259 api_timestamp = self.raw_json['api_timestamp'],
260 oar_version = self.raw_json['oar'] )
262 print self.version_json_dict['apilib_version']
265 def ParseTimezone(self) :
266 api_timestamp = self.raw_json['api_timestamp']
267 api_tz = self.raw_json['timezone']
268 return api_timestamp, api_tz
270 def ParseJobs(self) :
275 def ParseJobsTable(self) :
276 print "ParseJobsTable"
278 def ParseJobsDetails (self):
279 # currently, this function is not used a lot,
280 #so i have no idea what be usefull to parse,
281 #returning the full json. NT
282 #logger.debug("ParseJobsDetails %s " %(self.raw_json))
286 def ParseJobsIds(self):
288 job_resources = ['wanted_resources', 'name', 'id', 'start_time', \
289 'state','owner','walltime','message']
292 job_resources_full = ['launching_directory', 'links', \
293 'resubmit_job_id', 'owner', 'events', 'message', \
294 'scheduled_start', 'id', 'array_id', 'exit_code', \
295 'properties', 'state','array_index', 'walltime', \
296 'type', 'initial_request', 'stop_time', 'project',\
297 'start_time', 'dependencies','api_timestamp','submission_time', \
298 'reservation', 'stdout_file', 'types', 'cpuset_name', \
299 'name', 'wanted_resources','queue','stderr_file','command']
302 job_info = self.raw_json
303 #logger.debug("OARESTAPI ParseJobsIds %s" %(self.raw_json))
306 for k in job_resources:
307 values.append(job_info[k])
308 return dict(zip(job_resources, values))
311 logger.log_exc("ParseJobsIds KeyError ")
314 def ParseJobsIdResources(self):
315 """ Parses the json produced by the request
316 /oarapi/jobs/id/resources.json.
317 Returns a list of oar node ids that are scheduled for the
322 for resource in self.raw_json['items']:
323 job_resources.append(resource['id'])
325 #logger.debug("OARESTAPI \tParseJobsIdResources %s" %(self.raw_json))
328 def ParseResources(self) :
329 """ Parses the json produced by a get_resources request on oar."""
331 #logger.debug("OARESTAPI \tParseResources " )
332 #resources are listed inside the 'items' list from the json
333 self.raw_json = self.raw_json['items']
336 def ParseReservedNodes(self):
337 """ Returns an array containing the list of the reserved nodes """
339 #resources are listed inside the 'items' list from the json
340 reservation_list = []
341 print "ParseReservedNodes_%s" %(self.raw_json['items'])
343 #Parse resources info
344 for json_element in self.raw_json['items']:
345 #In case it is a real reservation (not asap case)
346 if json_element['scheduled_start']:
347 job['t_from'] = json_element['scheduled_start']
348 job['t_until'] = int(json_element['scheduled_start']) + \
349 int(json_element['walltime'])
350 #Get resources id list for the job
351 job['resource_ids'] = \
352 [ node_dict['id'] for node_dict in json_element['resources'] ]
354 job['t_from'] = "As soon as possible"
355 job['t_until'] = "As soon as possible"
356 job['resource_ids'] = ["Undefined"]
359 job['state'] = json_element['state']
360 job['lease_id'] = json_element['id']
363 job['user'] = json_element['owner']
364 #logger.debug("OARRestapi \tParseReservedNodes job %s" %(job))
365 reservation_list.append(job)
368 return reservation_list
370 def ParseRunningJobs(self):
371 """ Gets the list of nodes currently in use from the attributes of the
375 logger.debug("OARESTAPI \tParseRunningJobs__________________________ ")
376 #resources are listed inside the 'items' list from the json
378 for job in self.raw_json['items']:
379 for node in job['nodes']:
380 nodes.append(node['network_address'])
385 def ParseDeleteJobs(self):
386 """ No need to parse anything in this function.A POST
387 is done to delete the job.
392 def ParseResourcesFull(self) :
393 """ This method is responsible for parsing all the attributes
394 of all the nodes returned by OAR when issuing a get resources full.
395 The information from the nodes and the sites are separated.
396 Updates the node_dictlist so that the dictionnary of the platform's
397 nodes is available afterwards.
400 logger.debug("OARRESTAPI ParseResourcesFull________________________ ")
401 #print self.raw_json[1]
402 #resources are listed inside the 'items' list from the json
403 if self.version_json_dict['apilib_version'] != "0.2.10" :
404 self.raw_json = self.raw_json['items']
407 return self.node_dictlist
409 def ParseResourcesFullSites(self) :
410 """ UNUSED. Originally used to get information from the sites.
411 ParseResourcesFull is used instead.
414 if self.version_json_dict['apilib_version'] != "0.2.10" :
415 self.raw_json = self.raw_json['items']
418 return self.site_dict
422 def ParseNodes(self):
423 """ Parse nodes properties from OAR
424 Put them into a dictionary with key = node id and value is a dictionary
425 of the node properties and properties'values.
429 keys = self.resources_fulljson_dict.keys()
432 for dictline in self.raw_json:
434 # dictionary is empty and/or a new node has to be inserted
435 node_id = self.resources_fulljson_dict['network_address'](\
436 self.node_dictlist, dictline['network_address'])
439 if k == 'network_address':
442 self.resources_fulljson_dict[k](\
443 self.node_dictlist[node_id], dictline[k])
445 #The last property has been inserted in the property tuple list,
447 #Turn the property tuple list (=dict value) into a dictionary
448 self.node_dictlist[node_id] = dict(self.node_dictlist[node_id])
451 def slab_hostname_to_hrn(self, root_auth, hostname):
452 return root_auth + '.'+ hostname
456 def ParseSites(self):
457 """ Returns a list of dictionnaries containing the sites' attributes."""
461 #logger.debug(" OARrestapi.py \tParseSites self.node_dictlist %s"\
462 #%(self.node_dictlist))
463 # Create a list of nodes per site_id
464 for node_id in self.node_dictlist:
465 node = self.node_dictlist[node_id]
467 if node['site'] not in nodes_per_site:
468 nodes_per_site[node['site']] = []
469 nodes_per_site[node['site']].append(node['node_id'])
471 if node['node_id'] not in nodes_per_site[node['site']]:
472 nodes_per_site[node['site']].append(node['node_id'])
474 #Create a site dictionary whose key is site_login_base (name of the site)
475 # and value is a dictionary of properties, including the list
477 for node_id in self.node_dictlist:
478 node = self.node_dictlist[node_id]
479 #node.update({'hrn':self.slab_hostname_to_hrn(self.interface_hrn, \
480 #node['site'],node['hostname'])})
481 node.update({'hrn':self.slab_hostname_to_hrn(self.interface_hrn, node['hostname'])})
482 self.node_dictlist.update({node_id:node})
484 if node['site'] not in self.site_dict:
485 self.site_dict[node['site']] = {
487 'node_ids':nodes_per_site[node['site']],
488 'latitude':"48.83726",
489 'longitude':"- 2.10336",'name':config.SFA_REGISTRY_ROOT_AUTH,
490 'pcu_ids':[], 'max_slices':None, 'ext_consortium_id':None,
491 'max_slivers':None, 'is_public':True, 'peer_site_id': None,
492 'abbreviated_name':"senslab", 'address_ids': [],
493 'url':"http,//www.senslab.info", 'person_ids':[],
494 'site_tag_ids':[], 'enabled': True, 'slice_ids':[],
495 'date_created': None, 'peer_id': None }
496 #if node['site_login_base'] not in self.site_dict.keys():
497 #self.site_dict[node['site_login_base']] = {'login_base':node['site_login_base'],
498 #'node_ids':nodes_per_site[node['site_login_base']],
499 #'latitude':"48.83726",
500 #'longitude':"- 2.10336",'name':"senslab",
501 #'pcu_ids':[], 'max_slices':None, 'ext_consortium_id':None,
502 #'max_slivers':None, 'is_public':True, 'peer_site_id': None,
503 #'abbreviated_name':"senslab", 'address_ids': [],
504 #'url':"http,//www.senslab.info", 'person_ids':[],
505 #'site_tag_ids':[], 'enabled': True, 'slice_ids':[],
506 #'date_created': None, 'peer_id': None }
511 OARrequests_uri_dict = {
513 {'uri':'/oarapi/version.json', 'parse_func': ParseVersion},
515 {'uri':'/oarapi/timezone.json' ,'parse_func': ParseTimezone },
517 {'uri':'/oarapi/jobs.json','parse_func': ParseJobs},
519 {'uri':'/oarapi/jobs/id.json','parse_func': ParseJobsIds},
520 'GET_jobs_id_resources':
521 {'uri':'/oarapi/jobs/id/resources.json',\
522 'parse_func': ParseJobsIdResources},
524 {'uri':'/oarapi/jobs/table.json','parse_func': ParseJobsTable},
526 {'uri':'/oarapi/jobs/details.json',\
527 'parse_func': ParseJobsDetails},
528 'GET_reserved_nodes':
530 '/oarapi/jobs/details.json?state=Running,Waiting,Launching',\
532 'parse_func':ParseReservedNodes},
536 {'uri':'/oarapi/jobs/details.json?state=Running',\
537 'parse_func':ParseRunningJobs},
538 'GET_resources_full':
539 {'uri':'/oarapi/resources/full.json',\
540 'parse_func': ParseResourcesFull},
542 {'uri':'/oarapi/resources/full.json',\
543 'parse_func': ParseResourcesFullSites},
545 {'uri':'/oarapi/resources.json' ,'parse_func': ParseResources},
547 {'uri':'/oarapi/jobs/id.json' ,'parse_func': ParseDeleteJobs}
551 def FindNextPage(self):
552 if "links" in self.raw_json:
553 for page in self.raw_json['links']:
554 if page['rel'] == 'next':
555 self.concatenate = True
556 print>>sys.stderr, " \r\n \t\t FindNextPage self.concatenate %s" %(self.concatenate )
557 return True, "?"+page['href'].split("?")[1]
558 if self.concatenate :
560 print>>sys.stderr, " \r\n \t\t END FindNextPage self.concatenate %s" %(self.concatenate )
563 def ConcatenateJsonPages (self, saved_json_list):
568 print >>sys.stderr, " \r\n ConcatenateJsonPages saved_json_list len ", len(saved_json_list)
569 for page in saved_json_list:
570 #for node in page['items']:
571 #self.raw_json['items'].append(node)
572 print>>sys.stderr, " \r\n ConcatenateJsonPages page['items']len ", len(page['items'])
573 tmp['items'].extend(page['items'])
574 #print>>sys.stderr, " \r\n ConcatenateJsonPages len ", len(self.raw_json['items'])
575 #print>>sys.stderr, " \r\n ConcatenateJsonPages self.raw_json['items']", self.raw_json['items']
578 def SendRequest(self, request, strval = None , username = None):
579 """ Connects to OAR , sends the valid GET requests and uses
580 the appropriate json parsing functions.
587 self.concatenate = False
591 self.raw_json_list = []
592 if request in self.OARrequests_uri_dict :
594 self.raw_json = self.server.GETRequestToOARRestAPI(request, \
599 next_page , next_offset = self.FindNextPage()
601 #self.raw_json_list.append(self.raw_json)
602 save_json.append(self.raw_json)
603 if self.concatenate and self.end :
604 #self.raw_json = self.ConcatenateJsonPages(self.raw_json_list)
605 self.raw_json = self.ConcatenateJsonPages(save_json)
607 return self.OARrequests_uri_dict[request]['parse_func'](self)
609 logger.error("OARRESTAPI OARGetParse __init__ : ERROR_REQUEST " \