ViewVC Help

View File | Revision Log | Show Annotations | Root Listing

root/cvsroot/COMP/CRAB/python/DBinterface.py

Comparing COMP/CRAB/python/DBinterface.py (file contents):
Revision 1.20 by spiga, Tue Apr 22 10:37:00 2008 UTC vs.
Revision 1.72 by lolass, Mon Jan 21 11:22:46 2013 UTC

#	Line 1 | Line 1
1	–	from crab_logger import Logger
1		from crab_exceptions import *
2		from crab_util import *
3		import common
#	Line 34 | Line 33 | class DBinterface:
33		raise CrabException('Istantiate DB Session : '+str(e))
34
35		try:
36	<	common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
36	>	common.bossSession.bossLiteDB.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
37		except Exception, e :
38		raise CrabException('DB Installation error : '+str(e))
39		return
#	Line 56 | Line 55 | class DBinterface:
55		Return task with all/list of jobs
56		"""
57		try:
58	<	task = common.bossSession.load(1,jobsList)[0]
58	>	task = common.bossSession.load(1,jobsList)
59		except Exception, e :
60	<	common.logger.debug(3, "Error while getting task : " +str(traceback.format_exc()))
60	>	common.logger.debug( "Error while getting task : " +str(traceback.format_exc()))
61		raise CrabException('Error while getting task '+str(e))
62		return task
63
#	Line 67 | Line 66 | class DBinterface:
66		Return a task with a single job
67		"""
68		try:
69	<	task = common.bossSession.load(1,str(n))[0]
69	>	task = common.bossSession.load(1,str(n))
70		except Exception, e :
71	<	common.logger.debug(3, "Error while getting job : " +str(traceback.format_exc()))
71	>	common.logger.debug( "Error while getting job : " +str(traceback.format_exc()))
72		raise CrabException('Error while getting job '+str(e))
73		return task
74
#	Line 81 | Line 80 | class DBinterface:
80		"""
81		opt={}
82		if optsToSave.get('server_mode',0) == 1: opt['serverName']=optsToSave['server_name']
83	<	opt['name']=common.work_space.taskName()
84	<	task = Task( opt )
83	>	if common.scheduler.name().upper() not in ['LSF', 'CAF', 'SGE', 'PBS']:
84	>	checkNewSiteDB()
85	>	opt['name']= getUserName()+ '_' + string.split(common.work_space.topDir(),'/')[-2]+'_'+common.work_space.task_uuid()
86	>	task = Task( opt )
87		try:
88		common.bossSession.saveTask( task )
89		except Exception, e :
89	–	# common.logger.debug(3, "Error creating task : " +str(traceback.format_exc()))
90	–	# raise CrabException('Error creating task '+str(e))
90		raise CrabException('Error creating task '+str(traceback.format_exc()))
91
92		return
#	Line 107 | Line 106 | class DBinterface:
106
107		return
108
109	<	def createJobs_(self, jobsL):
109	>	def createJobs_(self, jobsL, isNew=True):
110		"""
111		Fill crab DB with  the jobs filed
112		"""
#	Line 116 | Line 115 | class DBinterface:
115		jobs = []
116		for id in jobsL:
117		parameters = {}
118	<	parameters['jobId'] =  str(id)
118	>	parameters['jobId'] = int(id)
119	>	parameters['taskId'] = 1
120		parameters['name'] = task['name'] + '_' + 'job' + str(id)
121		job = Job(parameters)
122		jobs.append(job)
123		common.bossSession.getRunningInstance(job)
124		job.runningJob['status'] = 'C'
125	<	task.addJobs(jobs)
125	>	## added to support second step creation
126	>	## maybe it is not needed. TO CLARIFY
127	>	if isNew:
128	>	task.addJobs(jobs)
129	>	else:
130	>	task.appendJobs(jobs)
131		try:
132		common.bossSession.updateDB( task )
133		except Exception, e :
#	Line 177 | Line 182 | class DBinterface:
182		"""
183		task = self.getTask(jobs)
184
180	–	Jobs = task.getJobs()
185		print "--------------------------"
186	<	for Job in Jobs:
187	<	print "Id: ",Job['id']
186	>	for Job in task.jobs:
187	>	print "Id: ",Job['jobId']
188		print "Dest: ", Job['dlsDestination']
189		print "Output: ", Job['outputFiles']
190		print "Args: ",Job['arguments']
191	+	print "Service: ",Job.runningJob['service']
192		print "--------------------------"
193		return
194
#	Line 192 | Line 197 | class DBinterface:
197		tmp_task = self.getTask()
198		return common.bossSession.serialize(tmp_task)
199
200	<	def queryID(self,server_mode=0):
200	>	def queryID(self,server_mode=0, jid=False):
201		'''
202		Return the taskId if serevr_mode =1
203		Return the joblistId if serevr_mode =0
#	Line 201 | Line 206 | class DBinterface:
206		lines=[]
207		task = self.getTask()
208		if server_mode == 1:
209	<	header= "Task Id = %-40s " %(task['name'])
210	<	else:
209	>	# init client server params...
210	>	CliServerParams(self)
211	>	headerTask = "Task Id = %-40s\n" %(task['name'])
212	>	headerTask+=  '--------------------------------------------------------------------------------------------\n'
213	>	displayReport(self,headerTask,lines)
214	>	common.logger.info(showWebMon(self.server_name))
215	>	if (jid ) or (server_mode == 0):
216		for job in task.jobs:
217		toPrint=''
218		common.bossSession.getRunningInstance(job)
219	<	toPrint = "%-5s %-50s " % (job['id'],job.runningJob['schedulerId'])
219	>	toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
220		lines.append(toPrint)
221	<	header+= "%-5s %-50s " % ('Job:','ID' )
222	<	displayReport(self,header,lines)
221	>	header+= "%-5s %-50s\n " % ('Job:','ID' )
222	>	header+=  '--------------------------------------------------------------------------------------------\n'
223	>	displayReport(self,header,lines)
224		return
225
226		def queryTask(self,attr):
#	Line 250 | Line 261 | class DBinterface:
261		try:
262		task = common.bossSession.loadJobDist( 1, attr )
263		except Exception, e :
264	<	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
264	>	common.logger.debug( "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
265		raise CrabException('Error loading Jobs By distinct Attr '+str(e))
266
267		for i in task: distAttr.append(i[attr])
#	Line 264 | Line 275 | class DBinterface:
275		try:
276		task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
277		except Exception, e :
278	<	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
278	>	common.logger.debug( "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
279		raise CrabException('Error loading Jobs By distinct Attr '+str(e))
280
281		for i in task: distAttr.append(i[attr_1])
#	Line 278 | Line 289 | class DBinterface:
289		try:
290		task = common.bossSession.loadJobsByAttr(attr )
291		except Exception, e :
292	<	common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
292	>	common.logger.debug( "Error loading Jobs By Attr : " +str(traceback.format_exc()))
293		raise CrabException('Error loading Jobs By Attr '+str(e))
294		for i in task:
295		matched.append(i[field])
#	Line 293 | Line 304 | class DBinterface:
304		try:
305		task = common.bossSession.loadJobsByRunningAttr(attr)
306		except Exception, e :
307	<	common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
307	>	common.logger.debug( "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
308		raise CrabException('Error loading Jobs By Running Attr '+str(e))
309		for i in task:
310		matched.append(i.runningJob[field])
311		return matched
312
313	+	def newRunJobs(self,nj='all'):
314	+	"""
315	+	Get new running instances
316	+	"""
317	+	task = self.getTask(nj)
318	+
319	+	for job in task.jobs:
320	+	common.bossSession.getNewRunningInstance(job)
321	+	job.runningJob['status'] = 'C'
322	+	job.runningJob['statusScheduler'] = 'Created'
323	+	job.runningJob['state'] = 'Created'
324	+	common.bossSession.updateDB(task)
325	+	return
326	+
327		def deserXmlStatus(self, reportList):
328
329		task = self.getTask()
330	<
330	>	if int(self.cfg_params.get('WMBS.automation',0)) == 1:
331	>	if len(reportList) ==0:
332	>	msg = 'You are using CRAB with WMBS the server is still creating your jobs.\n'
333	>	msg += '\tPlease wait...'
334	>	raise CrabException(msg)
335	>	newJobs =  len(reportList) - len(task.jobs)
336	>	if newJobs != 0:
337	>	isNew=True
338	>	if len(task.jobs):isNew=False
339	>	jobL=[]
340	>	for i in range(1,newJobs+1):
341	>	jobL.append(len(task.jobs)+i)
342	>	self.createJobs_(jobL,isNew)
343	>
344		for job in task.jobs:
345		if not job.runningJob:
346	<	raise CrabException( "Missing running object for job %s"%str(job['id']) )
346	>	raise CrabException( "Missing running object for job %s"%str(job['jobId']) )
347
348	<	id = str(job.runningJob['id'])
311	<	# TODO linear search, probably it can be optized with binary search
348	>	id = str(job.runningJob['jobId'])
349		rForJ = None
350	+	nj_list= []
351		for r in reportList:
352		if r.getAttribute('id') in [ id, 'all']:
353		rForJ = r
354		break
355
356	<	# Data alignment
357	<	jobStatus = str(job.runningJob['statusScheduler'])
358	<	if rForJ.getAttribute('status') not in ['Created', 'Submitting']:
359	<	job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
360	<	jobStatus = str(job.runningJob['statusScheduler'])
361	<	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
362	<
363	<	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
364	<	dest = str(job.runningJob['destination']).split(':')[0]
356	>	# check if rForJ is None
357	>	if rForJ is None:
358	>	common.logger.debug( "Missing XML element for job %s, skip update status"%str(id) )
359	>	continue
360	>
361	>	## Check the submission number and create new running jobs on the client side
362	>	if rForJ.getAttribute('resubmit') != 'None' and (rForJ.getAttribute('status') not in ['Cleared','Killed','Done','Done (Failed)','Not Submitted', 'Cancelled by user']) :
363	>	if int(job.runningJob['submission']) < int(rForJ.getAttribute('resubmit')) + 1:
364	>	nj_list.append(id)
365	>	if len(nj_list) > 0: self.newRunJobs(nj_list)
366
367	<	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
329	<	exe_exit_code = str(job.runningJob['applicationReturnCode'])
367	>	task_new = self.getTask()
368
369	<	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
370	<	job_exit_code = str(job.runningJob['wrapperReturnCode'])
369	>	for job in task_new.jobs:
370	>	id = str(job.runningJob['jobId'])
371	>	# TODO linear search, probably it can be optized with binary search
372	>	rForJ = None
373	>	for r in reportList:
374	>	if r.getAttribute('id') in [ id, 'all']:
375	>	rForJ = r
376	>	break
377	>
378	>	# Data alignment
379	>	if rForJ.getAttribute('status') not in ['Unknown']: # ['Created', 'Unknown']:
380	>	# update the status
381	>	common.logger.debug("Updating DB status for job: " + str(id) + " @: " \
382	>	+ str(rForJ.getAttribute('status')) )
383	>	job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
384	>	if (rForJ.getAttribute('status') == 'Done' or rForJ.getAttribute('status') == 'Done (Failed)')\
385	>	and rForJ.getAttribute('sched_status') == 'E' :
386	>	job.runningJob['status'] = 'SD'
387	>	else:
388	>	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
389	>
390	>	job.runningJob['schedulerId'] = str( rForJ.getAttribute('sched_id') )
391	>
392	>	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
393	>	dest = str(job.runningJob['destination']).split(':')[0]
394	>
395	>	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
396	>	exe_exit_code = str(job.runningJob['applicationReturnCode'])
397	>
398	>	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
399	>	job_exit_code = str(job.runningJob['wrapperReturnCode'])
400	>
401	>	job['closed'] = str( rForJ.getAttribute('ended') )
402	>
403	>	job.runningJob['state'] = str( rForJ.getAttribute('action') )
404	>
405	>	# Needed for unique naming of the output.
406	>	job['arguments'] = "%d %s"%(job.runningJob['jobId'], str(rForJ.getAttribute('submission')).strip() )
407	>
408	>	common.bossSession.updateDB( task_new )
409	>	return
410
411	<	if str( rForJ.getAttribute('resubmit') ).isdigit():
412	<	job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
413	<	# TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
411	>	# FIXME temporary method to verify what kind of submission to perform towards the server
412	>	def checkIfNeverSubmittedBefore(self):
413	>	for j in self.getTask().jobs:
414	>	if j.runningJob['submission'] > 1 or j.runningJob['state'] != 'Created':
415	>	return False
416	>	return True
417
418	<	common.bossSession.updateDB( task )
418	>	# Method to update arguments w.r.t. resubmission number in order to grant unique output
419	>	def updateResubAttribs(self, jobsL):
420	>	task = self.getTask(jobsL)
421	>	for j in task.jobs:
422	>	common.bossSession.getRunningInstance(j)
423	>	try:
424	>	resubNum = int(str(j['arguments']).split(' ')[1]) + 1
425	>	except Exception, e:
426	>	resubNum = j.runningJob['submission']
427	>	newArgs = "%d %d"%(j.runningJob['jobId'], resubNum)
428	>	j['arguments'] = newArgs
429
430	+	common.bossSession.updateDB(task)
431		return
432	+

Diff Legend

–	Removed lines
+	Added lines
<	Changed lines
>	Changed lines