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.1 by spiga, Mon Mar 3 20:14:43 2008 UTC vs.
Revision 1.49 by spiga, Mon Nov 10 13:51:58 2008 UTC

#	Line 3 | Line 3 | from crab_exceptions import *
3		from crab_util import *
4		import common
5		import os, time, shutil
6	+	import traceback
7
8		from ProdCommon.BossLite.API.BossLiteAPI import BossLiteAPI
9	<
9	>	from ProdCommon.BossLite.Common.Exceptions import DbError
10	>	from ProdCommon.BossLite.Common.Exceptions import TaskError
11
12		from ProdCommon.BossLite.DbObjects.Job import Job
13		from ProdCommon.BossLite.DbObjects.Task import Task
#	Line 13 | Line 15 | from ProdCommon.BossLite.DbObjects.Runni
15
16
17		class DBinterface:
18	<	def __init__(self):
17	<
18	<	return
19	<
20	<	def configureDB(self, cfg_params):
18	>	def __init__(self, cfg_params):
19
20		self.cfg_params = cfg_params
21
22		self.db_type =  cfg_params.get("USER.use_db",'SQLite')
23	+	return
24	+
25	+
26	+	def configureDB(self):
27
28		dbname = common.work_space.shareDir()+'crabDB'
29		dbConfig = {'dbName':dbname
30		}
31	+	try:
32	+	common.bossSession = BossLiteAPI( self.db_type, dbConfig)
33	+	except Exception, e :
34	+	raise CrabException('Istantiate DB Session : '+str(e))
35	+
36	+	try:
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
41
42	<	common.bossSession = BossLiteAPI( self.db_type, dbConfig)
43	<
44	<	common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
45	<
42	>	def loadDB(self):
43	>
44	>	dbname = common.work_space.shareDir()+'crabDB'
45	>	dbConfig = {'dbName':dbname
46	>	}
47	>	try:
48	>	common.bossSession = BossLiteAPI( self.db_type, dbConfig)
49	>	except Exception, e :
50	>	raise CrabException('Istantiate DB Session : '+str(e))
51	>
52		return
53	+
54	+	def getTask(self, jobsList='all'):
55	+	"""
56	+	Return task with all/list of jobs
57	+	"""
58	+	try:
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))
63	+	return task
64	+
65	+	def getJob(self, n):
66	+	"""
67	+	Return a task with a single job
68	+	"""
69	+	try:
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))
74	+	return task
75
76
77		def createTask_(self, optsToSave):
78		"""
79		Task declaration
80		with the first coniguration stuff
41	–	{'server_name': 'crabas.lnl.infn.it/data1/cms/', '-scheduler': 'glite', '-jobtype': 'cmssw', '-server_mode': '0'}
42	–
81		"""
82		opt={}
83	<	if optsToSave['server_mode'] == 1: opt['serverName']=optsToSave['server_name']
84	<	opt['jobType']=optsToSave['jobtype']
47	<	opt[ 'name']=common.work_space.taskName()
83	>	if optsToSave.get('server_mode',0) == 1: opt['serverName']=optsToSave['server_name']
84	>	opt['name']=common.work_space.taskName()
85		task = Task( opt )
86	<
87	<	common.bossSession.saveTask( task )
88	<	#common.bossSession.updateDB( task )
86	>	try:
87	>	common.bossSession.saveTask( task )
88	>	except Exception, e :
89	>	# common.logger.debug(3, "Error creating task : " +str(traceback.format_exc()))
90	>	# raise CrabException('Error creating task '+str(e))
91	>	raise CrabException('Error creating task '+str(traceback.format_exc()))
92	>
93		return
94
95		def updateTask_(self,optsToSave):
96		"""
97		Update task fields
98		"""
99	<	task = common.bossSession.loadTaskByName(common.work_space.taskName() )
100	<
99	>	task = self.getTask()
100	>
101		for key in optsToSave.keys():
102		task[key] = optsToSave[key]
103	+	try:
104		common.bossSession.updateDB( task )
105	+	except Exception, e :
106	+	raise CrabException('Error updating task '+str(traceback.format_exc()))
107	+
108		return
109
110	<	def createJobs_(self, nj):
110	>	def createJobs_(self, jobsL, isNew=True):
111		"""
112		Fill crab DB with  the jobs filed
113		"""
114	<	task = common.bossSession.loadTaskByName(common.work_space.taskName())
114	>	task = self.getTask()
115	>
116		jobs = []
117	<	for id in range(nj):
117	>	for id in jobsL:
118		parameters = {}
119	<	parameters['name'] = 'job' + 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)
125	<	common.bossSession.updateDB( task )
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 :
135	>	raise CrabException('Error updating task '+str(traceback.format_exc()))
136	>
137		return
138
139	<	def updateJob_(self, nj, optsToSave):
139	>	def updateJob_(self, jobsL, optsToSave):
140		"""
141		Update Job fields
142		"""
143	<	task = common.bossSession.loadTaskByName( common.work_space.taskName())
144	<	jobs = common.bossSession.loadJob(task['id'],nj+1)
145	<	for key in optsToSave.keys():
146	<	jobs[key] = optsToSave[key]
147	<	common.bossSession.updateDB( jobs )
143	>	task = self.getTask(jobsL)
144	>	id =0
145	>	for job in task.jobs:
146	>	for key in optsToSave[id].keys():
147	>	job[key] = optsToSave[id][key]
148	>	id+=1
149	>	try:
150	>	common.bossSession.updateDB( task )
151	>	except Exception, e :
152	>	raise CrabException('Error updating task '+str(traceback.format_exc()))
153		return
154
155	<	def updateRunJob_(self, nj, optsToSave):
155	>	def updateRunJob_(self, jobsL, optsToSave):
156		"""
157		Update Running Job fields
158		"""
159	<	task = common.bossSession.loadTaskByName( common.work_space.taskName())
160	<	common.bossSession.getRunningInstance(task.jobs[nj])
161	<	for key in optsToSave.keys():
162	<	task.jobs[nj].runningJob[key] = optsToSave[key]
159	>	task = self.getTask(jobsL)
160	>
161	>	id=0
162	>	for job in task.jobs:
163	>	common.bossSession.getRunningInstance(job)
164	>	for key in optsToSave[id].keys():
165	>	job.runningJob[key] = optsToSave[id][key]
166	>	id+=1
167		common.bossSession.updateDB( task )
168		return
169
170	<	def nJobs(self):
171	<	task = common.bossSession.loadTaskByName( common.work_space.taskName())
172	<	return len(task.jobs)
170	>	def nJobs(self,list=''):
171	>
172	>	task = self.getTask()
173	>	listId=[]
174	>	if list == 'list':
175	>	for job in task.jobs:listId.append(int(job['jobId']))
176	>	return listId
177	>	else:
178	>	return len(task.jobs)
179
180		def dump(self,jobs):
181		"""
182		List a complete set of infos for a job/range of jobs
183		"""
184	<	task = common.bossSession.loadTaskByName( common.work_space.taskName())
184	>	task = self.getTask(jobs)
185
186	<	njobs = len(jobs)
187	<	lines=[]
188	<	header=''
189	<	#   ##query the DB asking the right infos for runningJobs  TODO  DS
190	<	#   for job in jobs:
191	<	#       ## here the query over runngJobs
192	<	#       pass
193	<
120	<
121	<	#   ##Define Header to show and Pass the query results,
122	<	#   ##  header and format to displayReport()   TODO  DS
123	<	#   if njobs == 1: plural = ''
124	<	#   else:          plural = 's'
125	<	#   header += 'Listing %d job%s:\n' % (njobs, plural)
126	<	#   header += ' :\n' % (---) ## TODO DS
127	<
128	<	#   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):
197	+	if tmp_task is None:
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
205		'''
206		header=''
207		lines=[]
208	+	task = self.getTask()
209		if server_mode == 1:
210	<	header= "Task Id = %-40s " %(common.work_space.taskName())
211	<	else:
212	<	task = common.bossSession.loadTaskByName(common.work_space.taskName() )
213	<	for i in range(len(task.job)):
214	<	common.bossSession.getRunningInstance(task.jobs[i])
215	<	lines.append(task.jobs[i].runningJob['schedulerId'])
216	<
217	<	header+= "Job: %-5s Id = %-40s: \n"
218	<	displayReport(header,lines)
210	>	headerTask= "Task Id = %-40s " %(task['name'])
211	>	displayReport(self,headerTask,lines)
212	>	if (jid ) or (server_mode == 0):
213	>	for job in task.jobs:
214	>	toPrint=''
215	>	common.bossSession.getRunningInstance(job)
216	>	toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
217	>	lines.append(toPrint)
218	>	header+= "%-5s %-50s " % ('Job:','ID' )
219	>	displayReport(self,header,lines)
220		return
221
222		def queryTask(self,attr):
223		'''
224		Perform a query over a generic task attribute
225		'''
226	<	task = common.bossSession.loadTaskByName( common.work_space.taskName() )
226	>	task = self.getTask()
227		return task[attr]
228
229	<	def queryJob(self, attr, njobs):
229	>	def queryJob(self, attr, jobsL):
230		'''
231		Perform a query for a range/all/single job
232		over a generic job attribute
233		'''
234		lines=[]
235	<	task = common.bossSession.loadTaskByName( common.work_space.taskName())
236	<	for i in njobs:
237	<	jobs = common.bossSession.loadJob(task['id'],i+1)
166	<	lines.append(task.jobs[i][attr])
235	>	task = self.getTask(jobsL)
236	>	for job in task.jobs:
237	>	lines.append(job[attr])
238		return lines
239
240	<	def queryRunJob(self, attr, jobs):
240	>	def queryRunJob(self, attr, jobsL):
241		'''
242		Perform a query for a range/all/single job
243		over a generic job attribute
244		'''
245		lines=[]
246	<	task = common.bossSession.loadTaskByName( common.work_space.taskName() )
247	<	for i in jobs:
248	<	common.bossSession.getRunningInstance(task.jobs[i])
249	<	lines.append(task.jobs[i].runningJob[attr])
246	>	task = self.getTask(jobsL)
247	>	for job in task.jobs:
248	>	common.bossSession.getRunningInstance(job)
249	>	lines.append(job.runningJob[attr])
250		return lines
251	+
252	+	def queryDistJob(self, attr):
253	+	'''
254	+	Returns the list of distinct value for a given job attributes
255	+	'''
256	+	distAttr=[]
257	+	try:
258	+	task = common.bossSession.loadJobDist( 1, attr )
259	+	except Exception, e :
260	+	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
261	+	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
262	+
263	+	for i in task: distAttr.append(i[attr])
264	+	return  distAttr
265	+
266	+	def queryDistJob_Attr(self, attr_1, attr_2, list):
267	+	'''
268	+	Returns the list of distinct value for a given job attribute
269	+	'''
270	+	distAttr=[]
271	+	try:
272	+	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
273	+	except Exception, e :
274	+	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
275	+	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
276	+
277	+	for i in task: distAttr.append(i[attr_1])
278	+	return  distAttr
279	+
280	+	def queryAttrJob(self, attr, field):
281	+	'''
282	+	Returns the list of jobs matching the given attribute
283	+	'''
284	+	matched=[]
285	+	try:
286	+	task = common.bossSession.loadJobsByAttr(attr )
287	+	except Exception, e :
288	+	common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
289	+	raise CrabException('Error loading Jobs By Attr '+str(e))
290	+	for i in task:
291	+	matched.append(i[field])
292	+	return  matched
293	+
294	+
295	+	def queryAttrRunJob(self, attr,field):
296	+	'''
297	+	Returns the list of jobs matching the given attribute
298	+	'''
299	+	matched=[]
300	+	try:
301	+	task = common.bossSession.loadJobsByRunningAttr(attr)
302	+	except Exception, e :
303	+	common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
304	+	raise CrabException('Error loading Jobs By Running Attr '+str(e))
305	+	for i in task:
306	+	matched.append(i.runningJob[field])
307	+	return matched
308	+
309	+	def newRunJobs(self,nj='all'):
310	+	"""
311	+	Get new running instances
312	+	"""
313	+	task = self.getTask(nj)
314	+
315	+	for job in task.jobs:
316	+	common.bossSession.getNewRunningInstance(job)
317	+	job.runningJob['status'] = 'C'
318	+	job.runningJob['statusScheduler'] = 'Created'
319	+	common.bossSession.updateDB(task)
320	+	return
321	+
322	+	def deserXmlStatus(self, reportList):
323	+
324	+	task = self.getTask()
325	+	for job in task.jobs:
326	+	if not job.runningJob:
327	+	raise CrabException( "Missing running object for job %s"%str(job['jobId']) )
328	+
329	+	id = str(job.runningJob['jobId'])
330	+	rForJ = None
331	+	nj_list= []
332	+	for r in reportList:
333	+	if r.getAttribute('id') in [ id, 'all']:
334	+	rForJ = r
335	+	break
336	+	## Check the submission number and create new running jobs on the client side
337	+	if rForJ.getAttribute('resubmit') != 'None' and (rForJ.getAttribute('status') not in ['Cleared','Killed','Done','Done (Failed)']) :
338	+	if int(job.runningJob['submission']) < int(rForJ.getAttribute('resubmit')) + 1:
339	+	nj_list.append(id)
340	+	if len(nj_list) > 0: self.newRunJobs(nj_list)
341	+
342	+	task_new = self.getTask()
343	+
344	+	for job in task_new.jobs:
345	+	id = str(job.runningJob['jobId'])
346	+	# TODO linear search, probably it can be optized with binary search
347	+	rForJ = None
348	+	for r in reportList:
349	+	if r.getAttribute('id') in [ id, 'all']:
350	+	rForJ = r
351	+	break
352	+
353	+	# Data alignment
354	+	if rForJ.getAttribute('status') not in ['Created', 'Unknown']:
355	+	# update the status
356	+	common.logger.debug(3,"Updating DB status for job: " + str(id) + " @: " \
357	+	+ str(rForJ.getAttribute('status')) )
358	+	job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
359	+	if (rForJ.getAttribute('status') == 'Done' or rForJ.getAttribute('status') == 'Done (Failed)')\
360	+	and rForJ.getAttribute('sched_status') == 'E' :
361	+	job.runningJob['status'] = 'SD'
362	+	else:
363	+	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
364	+
365	+	job.runningJob['schedulerId'] = str( rForJ.getAttribute('sched_id') )
366	+
367	+	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
368	+	dest = str(job.runningJob['destination']).split(':')[0]
369	+
370	+	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
371	+	exe_exit_code = str(job.runningJob['applicationReturnCode'])
372	+
373	+	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
374	+	job_exit_code = str(job.runningJob['wrapperReturnCode'])
375	+
376	+	#if str( rForJ.getAttribute('resubmit') ).isdigit():
377	+	#    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
378	+	#    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
379	+
380	+	# TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
381	+
382	+	common.bossSession.updateDB( task_new )
383	+	return
384	+
385	+	# FIXME temporary method to verify what kind of submission to perform towards the server
386	+	def checkIfNeverSubmittedBefore(self):
387	+	for j in self.getTask().jobs:
388	+	if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C' or \
389	+	j.runningJob['statusScheduler'] != 'Created':
390	+	return False
391	+	return True
392	+
393	+

Diff Legend

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