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.4 by spiga, Wed Mar 5 18:30:40 2008 UTC vs.
Revision 1.24 by spiga, Sat May 10 13:41:40 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 26 | Line 28 | class DBinterface:
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.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
38	+	except Exception, e :
39	+	raise CrabException('DB Installation error : '+str(e))
40	+	return
41
30	–	common.bossSession = BossLiteAPI( self.db_type, dbConfig)
31	–	common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
32	–
33	–	return
34	–
42		def loadDB(self):
43
44		dbname = common.work_space.shareDir()+'crabDB'
45		dbConfig = {'dbName':dbname
46		}
47	<	common.bossSession = BossLiteAPI( self.db_type, dbConfig)
48	<
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):
55	<
56	<	self.task = common.bossSession.loadTaskByID(1)
57	<	return self.task
58	<
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)[0]
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))[0]
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
55	–	{'server_name': 'crabas.lnl.infn.it/data1/cms/', '-scheduler': 'glite', '-jobtype': 'cmssw', '-server_mode': '0'}
56	–
81		"""
82		opt={}
83	<	if optsToSave['server_mode'] == 1: opt['serverName']=optsToSave['server_name']
84	<	opt['jobType']=optsToSave['jobtype']
61	<	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	<	task = common.bossSession.loadTaskByID(1)
74	<
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):
111		"""
112		Fill crab DB with  the jobs filed
113		"""
114	<	#task = common.bossSession.loadTaskByName(common.work_space.taskName())
115	<	task = common.bossSession.loadTaskByID(1)
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'] =  str(id)
120	>	parameters['name'] = task['name'] + '_' + 'job' + str(id)
121		job = Job(parameters)
122	<	jobs.append(job)
122	>	jobs.append(job)
123	>	common.bossSession.getRunningInstance(job)
124	>	job.runningJob['status'] = 'C'
125		task.addJobs(jobs)
126	<	common.bossSession.updateDB( task )
126	>	try:
127	>	common.bossSession.updateDB( task )
128	>	except Exception, e :
129	>	raise CrabException('Error updating task '+str(traceback.format_exc()))
130	>
131		return
132
133	<	def updateJob_(self, nj, optsToSave):
133	>	def updateJob_(self, jobsL, optsToSave):
134		"""
135		Update Job fields
136		"""
137	<	task = common.bossSession.loadTaskByID(1)
138	<	#task = common.bossSession.loadTaskByName( common.work_space.taskName())
139	<	jobs = common.bossSession.loadJob(task['id'],nj+1)
140	<	for key in optsToSave.keys():
141	<	jobs[key] = optsToSave[key]
142	<	common.bossSession.updateDB( jobs )
137	>	task = self.getTask(jobsL)
138	>	id =0
139	>	for job in task.jobs:
140	>	for key in optsToSave[id].keys():
141	>	job[key] = optsToSave[id][key]
142	>	id+=1
143	>	try:
144	>	common.bossSession.updateDB( task )
145	>	except Exception, e :
146	>	raise CrabException('Error updating task '+str(traceback.format_exc()))
147		return
148
149	<	def updateRunJob_(self, nj, optsToSave):
149	>	def updateRunJob_(self, jobsL, optsToSave):
150		"""
151		Update Running Job fields
152		"""
153	<	task = common.bossSession.loadTaskByID(1)
154	<	#task = common.bossSession.loadTaskByName( common.work_space.taskName())
155	<	common.bossSession.getRunningInstance(task.jobs[nj])
156	<	for key in optsToSave.keys():
157	<	task.jobs[nj].runningJob[key] = optsToSave[key]
153	>	task = self.getTask(jobsL)
154	>
155	>	id=0
156	>	for job in task.jobs:
157	>	common.bossSession.getRunningInstance(job)
158	>	for key in optsToSave[id].keys():
159	>	job.runningJob[key] = optsToSave[id][key]
160	>	id+=1
161		common.bossSession.updateDB( task )
162		return
163
164	<	def nJobs(self):
164	>	def nJobs(self,list=''):
165
166	<	task = common.bossSession.loadTaskByID(1)
167	<	#task = common.bossSession.loadTaskByName( common.work_space.taskName())
168	<	return len(task.jobs)
166	>	task = self.getTask()
167	>	listId=[]
168	>	if list == 'list':
169	>	for job in task.jobs:listId.append(int(job['jobId']))
170	>	return listId
171	>	else:
172	>	return len(task.jobs)
173
174		def dump(self,jobs):
175		"""
176		List a complete set of infos for a job/range of jobs
177		"""
178	<	task = common.bossSession.loadTaskByID(1)
131	<	#task = common.bossSession.loadTaskByName( common.work_space.taskName())
132	<
133	<	njobs = len(jobs)
134	<	lines=[]
135	<	header=''
136	<	#   ##query the DB asking the right infos for runningJobs  TODO  DS
137	<	#   for job in jobs:
138	<	#       ## here the query over runngJobs
139	<	#       pass
140	<
178	>	task = self.getTask(jobs)
179
180	<	#   ##Define Header to show and Pass the query results,
181	<	#   ##  header and format to displayReport()   TODO  DS
182	<	#   if njobs == 1: plural = ''
183	<	#   else:          plural = 's'
184	<	#   header += 'Listing %d job%s:\n' % (njobs, plural)
185	<	#   header += ' :\n' % (---) ## TODO DS
186	<
187	<	#   displayReport(header, lines):
180	>	Jobs = task.getJobs()
181	>	print "--------------------------"
182	>	for Job in Jobs:
183	>	print "Id: ",Job['jobId']
184	>	print "Dest: ", Job['dlsDestination']
185	>	print "Output: ", Job['outputFiles']
186	>	print "Args: ",Job['arguments']
187	>	print "--------------------------"
188		return
189	+
190	+	def serializeTask(self, tmp_task = None):
191	+	if tmp_task is None:
192	+	tmp_task = self.getTask()
193	+	return common.bossSession.serialize(tmp_task)
194
195		def queryID(self,server_mode=0):
196		'''
#	Line 156 | Line 199 | class DBinterface:
199		'''
200		header=''
201		lines=[]
202	<	task = common.bossSession.loadTaskByID(1)
202	>	task = self.getTask()
203		if server_mode == 1:
204		header= "Task Id = %-40s " %(task['name'])
205		else:
206	<	#   task = common.bossSession.loadTaskByName(common.work_space.taskName() )
207	<	for i in range(len(task.job)):
208	<	common.bossSession.getRunningInstance(task.jobs[i])
209	<	lines.append(task.jobs[i].runningJob['schedulerId'])
210	<
211	<	header+= "Job: %-5s Id = %-40s: \n"
212	<	displayReport(header,lines)
206	>	for job in task.jobs:
207	>	toPrint=''
208	>	common.bossSession.getRunningInstance(job)
209	>	toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
210	>	lines.append(toPrint)
211	>	header+= "%-5s %-50s " % ('Job:','ID' )
212	>	displayReport(self,header,lines)
213		return
214
215		def queryTask(self,attr):
216		'''
217		Perform a query over a generic task attribute
218		'''
219	<	task = common.bossSession.loadTaskByID(1)
219	>	task = self.getTask()
220		return task[attr]
221
222	<	def queryJob(self, attr, njobs):
222	>	def queryJob(self, attr, jobsL):
223		'''
224		Perform a query for a range/all/single job
225		over a generic job attribute
226		'''
227		lines=[]
228	<	task = common.bossSession.loadTaskByID(1)
229	<	#task = common.bossSession.loadTaskByName( common.work_space.taskName())
230	<	for i in njobs:
188	<	jobs = common.bossSession.loadJob(task['id'],i+1)
189	<	lines.append(task.jobs[i][attr])
228	>	task = self.getTask(jobsL)
229	>	for job in task.jobs:
230	>	lines.append(job[attr])
231		return lines
232
233	<	def queryRunJob(self, attr, jobs):
233	>	def queryRunJob(self, attr, jobsL):
234		'''
235		Perform a query for a range/all/single job
236		over a generic job attribute
237		'''
238		lines=[]
239	<	task = common.bossSession.loadTaskByID(1)
240	<	# task = common.bossSession.loadTaskByName( common.work_space.taskName() )
241	<	for i in jobs:
242	<	common.bossSession.getRunningInstance(task.jobs[i])
202	<	lines.append(task.jobs[i].runningJob[attr])
239	>	task = self.getTask(jobsL)
240	>	for job in task.jobs:
241	>	common.bossSession.getRunningInstance(job)
242	>	lines.append(job.runningJob[attr])
243		return lines
244
245		def queryDistJob(self, attr):
#	Line 207 | Line 247 | class DBinterface:
247		Returns the list of distinct value for a given job attributes
248		'''
249		distAttr=[]
250	<	task = common.bossSession.loadJobDistAttr( 1, attr )
250	>	try:
251	>	task = common.bossSession.loadJobDist( 1, attr )
252	>	except Exception, e :
253	>	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
254	>	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
255	>
256		for i in task: distAttr.append(i[attr])
257		return  distAttr
258
259	<	def queryDistJobHadr(self, attr_1, attr_2, list):
259	>	def queryDistJob_Attr(self, attr_1, attr_2, list):
260		'''
261	<	Returns the list of distinct value for a given job attributes
261	>	Returns the list of distinct value for a given job attribute
262		'''
263		distAttr=[]
264	<	task = common.bossSession.loadJobDistAttrHadr( 1, attr_1, attr_2, list )
265	<	for i in task: distAttr.append(i[attr_1])
264	>	try:
265	>	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
266	>	except Exception, e :
267	>	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
268	>	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
269	>
270	>	for i in task: distAttr.append(i[attr_1])
271		return  distAttr
272	+
273		def queryAttrJob(self, attr, field):
274		'''
275		Returns the list of jobs matching the given attribute
276		'''
277		matched=[]
278	<	task = common.bossSession.loadJobsByAttr(attr )
278	>	try:
279	>	task = common.bossSession.loadJobsByAttr(attr )
280	>	except Exception, e :
281	>	common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
282	>	raise CrabException('Error loading Jobs By Attr '+str(e))
283		for i in task:
284		matched.append(i[field])
285		return  matched
286
287	+
288	+	def queryAttrRunJob(self, attr,field):
289	+	'''
290	+	Returns the list of jobs matching the given attribute
291	+	'''
292	+	matched=[]
293	+	try:
294	+	task = common.bossSession.loadJobsByRunningAttr(attr)
295	+	except Exception, e :
296	+	common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
297	+	raise CrabException('Error loading Jobs By Running Attr '+str(e))
298	+	for i in task:
299	+	matched.append(i.runningJob[field])
300	+	return matched
301	+
302	+	def newRunJobs(self,nj='all'):
303	+	"""
304	+	Get new running instances
305	+	"""
306	+	task = self.getTask(nj)
307	+
308	+	for job in task.jobs:
309	+	common.bossSession.getNewRunningInstance(job)
310	+	job.runningJob['status'] = 'C'
311	+	job.runningJob['statusScheduler'] = 'Created'
312	+	common.bossSession.updateDB(task)
313	+	return
314	+
315	+	def deserXmlStatus(self, reportList):
316	+
317	+	task = self.getTask()
318	+
319	+	for job in task.jobs:
320	+	if not job.runningJob:
321	+	raise CrabException( "Missing running object for job %s"%str(job['id']) )
322	+
323	+	id = str(job.runningJob['id'])
324	+	# TODO linear search, probably it can be optized with binary search
325	+	rForJ = None
326	+	for r in reportList:
327	+	if r.getAttribute('id') in [ id, 'all']:
328	+	rForJ = r
329	+	break
330	+
331	+	# Data alignment
332	+	jobStatus = str(job.runningJob['statusScheduler'])
333	+	if rForJ.getAttribute('status') not in ['Created', 'Submitting', 'Unknown'] and \
334	+	job.runningJob['statusScheduler'] != 'Cleared':
335	+	job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
336	+	jobStatus = str(job.runningJob['statusScheduler'])
337	+	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
338	+
339	+	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
340	+	dest = str(job.runningJob['destination']).split(':')[0]
341	+
342	+	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
343	+	exe_exit_code = str(job.runningJob['applicationReturnCode'])
344	+
345	+	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
346	+	job_exit_code = str(job.runningJob['wrapperReturnCode'])
347	+
348	+	#if str( rForJ.getAttribute('resubmit') ).isdigit():
349	+	#    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
350	+	#    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
351	+
352	+	# TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
353	+
354	+	common.bossSession.updateDB( task )
355	+
356	+	return

Diff Legend

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