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.15 by mcinquil, Tue Apr 15 14:13:07 2008 UTC vs.
Revision 1.54 by spiga, Sat Mar 7 16:32:14 2009 UTC

#	Line 34 | Line 34 | class DBinterface:
34		raise CrabException('Istantiate DB Session : '+str(e))
35
36		try:
37	<	common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
37	>	common.bossSession.bossLiteDB.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
38		except Exception, e :
39		raise CrabException('DB Installation error : '+str(e))
40		return
#	Line 56 | Line 56 | class DBinterface:
56		Return task with all/list of jobs
57		"""
58		try:
59	<	task = common.bossSession.load(1,jobsList)[0]
59	>	task = common.bossSession.load(1,jobsList)
60		except Exception, e :
61		common.logger.debug(3, "Error while getting task : " +str(traceback.format_exc()))
62		raise CrabException('Error while getting task '+str(e))
#	Line 67 | Line 67 | class DBinterface:
67		Return a task with a single job
68		"""
69		try:
70	<	task = common.bossSession.load(1,str(n))[0]
70	>	task = common.bossSession.load(1,str(n))
71		except Exception, e :
72		common.logger.debug(3, "Error while getting job : " +str(traceback.format_exc()))
73		raise CrabException('Error while getting job '+str(e))
#	Line 81 | Line 81 | class DBinterface:
81		"""
82		opt={}
83		if optsToSave.get('server_mode',0) == 1: opt['serverName']=optsToSave['server_name']
84	<	opt['name']=common.work_space.taskName()
84	>	opt['name']= getUserName(self)+ '_' + string.split(common.work_space.topDir(),'/')[-2]+'_'+common.work_space.task_uuid()
85		task = Task( opt )
86		try:
87		common.bossSession.saveTask( task )
#	Line 107 | Line 107 | class DBinterface:
107
108		return
109
110	<	def createJobs_(self, jobsL):
110	>	def createJobs_(self, jobsL, isNew=True):
111		"""
112		Fill crab DB with  the jobs filed
113		"""
#	Line 116 | Line 116 | class DBinterface:
116		jobs = []
117		for id in jobsL:
118		parameters = {}
119	<	parameters['jobId'] =  str(id)
119	>	parameters['jobId'] = int(id)
120	>	parameters['taskId'] = 1
121		parameters['name'] = task['name'] + '_' + 'job' + str(id)
122		job = Job(parameters)
123	<	jobs.append(job)
124	<	task.addJobs(jobs)
123	>	jobs.append(job)
124	>	common.bossSession.getRunningInstance(job)
125	>	job.runningJob['status'] = 'C'
126	>	## added to support second step creation
127	>	## maybe it is not needed. TO CLARIFY
128	>	if isNew:
129	>	task.addJobs(jobs)
130	>	else:
131	>	task.appendJobs(jobs)
132		try:
133		common.bossSession.updateDB( task )
134		except Exception, e :
#	Line 173 | Line 181 | class DBinterface:
181		"""
182		List a complete set of infos for a job/range of jobs
183		"""
184	<	task = self.getTask()
177	<
178	<	njobs = len(jobs)
179	<	lines=[]
180	<	header=''
181	<	#   ##query the DB asking the right infos for runningJobs  TODO  DS
182	<	#   for job in jobs:
183	<	#       ## here the query over runngJobs
184	<	#       pass
185	<
186	<
187	<	#   ##Define Header to show and Pass the query results,
188	<	#   ##  header and format to displayReport()   TODO  DS
189	<	#   if njobs == 1: plural = ''
190	<	#   else:          plural = 's'
191	<	#   header += 'Listing %d job%s:\n' % (njobs, plural)
192	<	#   header += ' :\n' % (---) ## TODO DS
184	>	task = self.getTask(jobs)
185
186	<	#   displayReport(header, lines):
186	>	print "--------------------------"
187	>	for Job in task.jobs:
188	>	print "Id: ",Job['jobId']
189	>	print "Dest: ", Job['dlsDestination']
190	>	print "Output: ", Job['outputFiles']
191	>	print "Args: ",Job['arguments']
192	>	print "Service: ",Job.runningJob['service']
193	>	print "--------------------------"
194		return
195
196		def serializeTask(self, tmp_task = None):
#	Line 199 | Line 198 | class DBinterface:
198		tmp_task = self.getTask()
199		return common.bossSession.serialize(tmp_task)
200
201	<	def queryID(self,server_mode=0):
201	>	def queryID(self,server_mode=0, jid=False):
202		'''
203		Return the taskId if serevr_mode =1
204		Return the joblistId if serevr_mode =0
#	Line 208 | Line 207 | class DBinterface:
207		lines=[]
208		task = self.getTask()
209		if server_mode == 1:
210	<	header= "Task Id = %-40s " %(task['name'])
211	<	else:
210	>	# init client server params...
211	>	CliServerParams(self)
212	>	headerTask= "Task Id = %-40s " %(task['name'])
213	>	displayReport(self,headerTask,lines)
214	>	common.logger.message(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)
222	>	displayReport(self,header,lines)
223		return
224
225		def queryTask(self,attr):
#	Line 234 | Line 237 | class DBinterface:
237		lines=[]
238		task = self.getTask(jobsL)
239		for job in task.jobs:
240	<	lines.append(eval(job[attr]))
240	>	lines.append(job[attr])
241		return lines
242
243		def queryRunJob(self, attr, jobsL):
#	Line 260 | Line 263 | class DBinterface:
263		common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
264		raise CrabException('Error loading Jobs By distinct Attr '+str(e))
265
266	<	for i in task: distAttr.append(eval(i[attr]))
266	>	for i in task: distAttr.append(i[attr])
267		return  distAttr
268
269		def queryDistJob_Attr(self, attr_1, attr_2, list):
#	Line 274 | Line 277 | class DBinterface:
277		common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
278		raise CrabException('Error loading Jobs By distinct Attr '+str(e))
279
280	<	for i in task: distAttr.append(eval(i[attr_1]))
280	>	for i in task: distAttr.append(i[attr_1])
281		return  distAttr
282
283		def queryAttrJob(self, attr, field):
#	Line 303 | Line 306 | class DBinterface:
306		common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
307		raise CrabException('Error loading Jobs By Running Attr '+str(e))
308		for i in task:
309	<	matched.append(i[field])
309	>	matched.append(i.runningJob[field])
310		return matched
311	+
312	+	def newRunJobs(self,nj='all'):
313	+	"""
314	+	Get new running instances
315	+	"""
316	+	task = self.getTask(nj)
317	+
318	+	for job in task.jobs:
319	+	common.bossSession.getNewRunningInstance(job)
320	+	job.runningJob['status'] = 'C'
321	+	job.runningJob['statusScheduler'] = 'Created'
322	+	common.bossSession.updateDB(task)
323	+	return
324	+
325	+	def deserXmlStatus(self, reportList):
326	+
327	+	task = self.getTask()
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'])
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']) :
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	+	if rForJ.getAttribute('status') not in ['Created', 'Unknown']:
358	+	# update the status
359	+	common.logger.debug(3,"Updating DB status for job: " + str(id) + " @: " \
360	+	+ str(rForJ.getAttribute('status')) )
361	+	job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
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	+	## unsing 'standardInput' field for 'ended' tag ['Y','N']
380	+	job['standardInput'] = str( rForJ.getAttribute('ended') )
381	+
382	+	#if str( rForJ.getAttribute('resubmit') ).isdigit():
383	+	#    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
384	+	#    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
385	+
386	+	# TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
387	+
388	+	common.bossSession.updateDB( task_new )
389	+	return
390	+
391	+	# FIXME temporary method to verify what kind of submission to perform towards the server
392	+	def checkIfNeverSubmittedBefore(self):
393	+	for j in self.getTask().jobs:
394	+	if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C' or \
395	+	j.runningJob['statusScheduler'] != 'Created':
396	+	return False
397	+	return True
398	+
399	+

Diff Legend

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