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.30 by spiga, Tue May 20 20:49:11 2008 UTC vs.
Revision 1.63 by mcinquil, Thu Jul 23 08:57:37 2009 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 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()
83	>	opt['name']= getUserName()+ '_' + string.split(common.work_space.topDir(),'/')[-2]+'_'+common.work_space.task_uuid()
84		task = Task( opt )
85		try:
86		common.bossSession.saveTask( task )
87		except Exception, e :
89	–	# common.logger.debug(3, "Error creating task : " +str(traceback.format_exc()))
90	–	# raise CrabException('Error creating task '+str(e))
88		raise CrabException('Error creating task '+str(traceback.format_exc()))
89
90		return
#	Line 107 | Line 104 | class DBinterface:
104
105		return
106
107	<	def createJobs_(self, jobsL):
107	>	def createJobs_(self, jobsL, isNew=True):
108		"""
109		Fill crab DB with  the jobs filed
110		"""
#	Line 116 | Line 113 | class DBinterface:
113		jobs = []
114		for id in jobsL:
115		parameters = {}
116	<	parameters['jobId'] =  str(id)
116	>	parameters['jobId'] = int(id)
117	>	parameters['taskId'] = 1
118		parameters['name'] = task['name'] + '_' + 'job' + str(id)
119		job = Job(parameters)
120		jobs.append(job)
121		common.bossSession.getRunningInstance(job)
122		job.runningJob['status'] = 'C'
123	<	task.addJobs(jobs)
123	>	## added to support second step creation
124	>	## maybe it is not needed. TO CLARIFY
125	>	if isNew:
126	>	task.addJobs(jobs)
127	>	else:
128	>	task.appendJobs(jobs)
129		try:
130		common.bossSession.updateDB( task )
131		except Exception, e :
#	Line 177 | Line 180 | class DBinterface:
180		"""
181		task = self.getTask(jobs)
182
180	–	Jobs = task.getJobs()
183		print "--------------------------"
184	<	for Job in Jobs:
184	>	for Job in task.jobs:
185		print "Id: ",Job['jobId']
186		print "Dest: ", Job['dlsDestination']
187		print "Output: ", Job['outputFiles']
188		print "Args: ",Job['arguments']
189	+	print "Service: ",Job.runningJob['service']
190		print "--------------------------"
191		return
192
#	Line 192 | Line 195 | class DBinterface:
195		tmp_task = self.getTask()
196		return common.bossSession.serialize(tmp_task)
197
198	<	def queryID(self,server_mode=0):
198	>	def queryID(self,server_mode=0, jid=False):
199		'''
200		Return the taskId if serevr_mode =1
201		Return the joblistId if serevr_mode =0
#	Line 201 | Line 204 | class DBinterface:
204		lines=[]
205		task = self.getTask()
206		if server_mode == 1:
207	<	header= "Task Id = %-40s " %(task['name'])
208	<	else:
207	>	# init client server params...
208	>	CliServerParams(self)
209	>	headerTask = "Task Id = %-40s\n" %(task['name'])
210	>	headerTask+=  '--------------------------------------------------------------------------------------------\n'
211	>	displayReport(self,headerTask,lines)
212	>	common.logger.info(showWebMon(self.server_name))
213	>	if (jid ) or (server_mode == 0):
214		for job in task.jobs:
215		toPrint=''
216		common.bossSession.getRunningInstance(job)
217		toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
218		lines.append(toPrint)
219	<	header+= "%-5s %-50s " % ('Job:','ID' )
220	<	displayReport(self,header,lines)
219	>	header+= "%-5s %-50s\n " % ('Job:','ID' )
220	>	header+=  '--------------------------------------------------------------------------------------------\n'
221	>	displayReport(self,header,lines)
222		return
223
224		def queryTask(self,attr):
#	Line 250 | Line 259 | class DBinterface:
259		try:
260		task = common.bossSession.loadJobDist( 1, attr )
261		except Exception, e :
262	<	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
262	>	common.logger.debug( "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
263		raise CrabException('Error loading Jobs By distinct Attr '+str(e))
264
265		for i in task: distAttr.append(i[attr])
#	Line 264 | Line 273 | class DBinterface:
273		try:
274		task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
275		except Exception, e :
276	<	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
276	>	common.logger.debug( "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
277		raise CrabException('Error loading Jobs By distinct Attr '+str(e))
278
279		for i in task: distAttr.append(i[attr_1])
#	Line 278 | Line 287 | class DBinterface:
287		try:
288		task = common.bossSession.loadJobsByAttr(attr )
289		except Exception, e :
290	<	common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
290	>	common.logger.debug( "Error loading Jobs By Attr : " +str(traceback.format_exc()))
291		raise CrabException('Error loading Jobs By Attr '+str(e))
292		for i in task:
293		matched.append(i[field])
#	Line 293 | Line 302 | class DBinterface:
302		try:
303		task = common.bossSession.loadJobsByRunningAttr(attr)
304		except Exception, e :
305	<	common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
305	>	common.logger.debug( "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
306		raise CrabException('Error loading Jobs By Running Attr '+str(e))
307		for i in task:
308		matched.append(i.runningJob[field])
#	Line 309 | Line 318 | class DBinterface:
318		common.bossSession.getNewRunningInstance(job)
319		job.runningJob['status'] = 'C'
320		job.runningJob['statusScheduler'] = 'Created'
321	+	job.runningJob['state'] = 'Created'
322		common.bossSession.updateDB(task)
323		return
324
325		def deserXmlStatus(self, reportList):
326
327		task = self.getTask()
318	–
328		for job in task.jobs:
329		if not job.runningJob:
330		raise CrabException( "Missing running object for job %s"%str(job['jobId']) )
331
332		id = str(job.runningJob['jobId'])
324	–	# TODO linear search, probably it can be optized with binary search
333		rForJ = None
334	+	nj_list= []
335		for r in reportList:
336		if r.getAttribute('id') in [ id, 'all']:
337		rForJ = r
338		break
339	+	## Check the submission number and create new running jobs on the client side
340	+	if rForJ.getAttribute('resubmit') != 'None' and (rForJ.getAttribute('status') not in ['Cleared','Killed','Done','Done (Failed)','Not Submitted', 'Cancelled by user']) :
341	+	if int(job.runningJob['submission']) < int(rForJ.getAttribute('resubmit')) + 1:
342	+	nj_list.append(id)
343	+	if len(nj_list) > 0: self.newRunJobs(nj_list)
344	+
345	+	task_new = self.getTask()
346
347	+	for job in task_new.jobs:
348	+	id = str(job.runningJob['jobId'])
349	+	# TODO linear search, probably it can be optized with binary search
350	+	rForJ = None
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', 'Unknown'] and \
359	<	job.runningJob['statusScheduler'] != 'Cleared':
357	>	if rForJ.getAttribute('status') not in ['Unknown']: # ['Created', 'Unknown']:
358	>	# update the status
359	>	common.logger.debug("Updating DB status for job: " + str(id) + " @: " \
360	>	+ str(rForJ.getAttribute('status')) )
361		job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
362	<	jobStatus = str(job.runningJob['statusScheduler'])
363	<	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
364	<
365	<	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
366	<	dest = str(job.runningJob['destination']).split(':')[0]
367	<
368	<	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
369	<	exe_exit_code = str(job.runningJob['applicationReturnCode'])
362	>	if (rForJ.getAttribute('status') == 'Done' or rForJ.getAttribute('status') == 'Done (Failed)')\
363	>	and rForJ.getAttribute('sched_status') == 'E' :
364	>	job.runningJob['status'] = 'SD'
365	>	else:
366	>	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
367	>
368	>	job.runningJob['schedulerId'] = str( rForJ.getAttribute('sched_id') )
369	>
370	>	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
371	>	dest = str(job.runningJob['destination']).split(':')[0]
372	>
373	>	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
374	>	exe_exit_code = str(job.runningJob['applicationReturnCode'])
375	>
376	>	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
377	>	job_exit_code = str(job.runningJob['wrapperReturnCode'])
378
379	<	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
346	<	job_exit_code = str(job.runningJob['wrapperReturnCode'])
379	>	job['closed'] = str( rForJ.getAttribute('ended') )
380
381	+	job.runningJob['state'] = str( rForJ.getAttribute('action') )
382	+
383		#if str( rForJ.getAttribute('resubmit') ).isdigit():
384		#    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
385		#    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
386	<
386	>
387		# TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
388
389	<	common.bossSession.updateDB( task )
355	<
389	>	common.bossSession.updateDB( task_new )
390		return
391
392		# FIXME temporary method to verify what kind of submission to perform towards the server
393		def checkIfNeverSubmittedBefore(self):
394		for j in self.getTask().jobs:
395	<	if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C':
395	>	if j.runningJob['submission'] > 1 or j.runningJob['state'] != 'Created':
396		return False
397		return True
398

Diff Legend

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