CRAB/python/DBinterface.py

from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
import common
import os, time, shutil
import traceback

from ProdCommon.BossLite.API.BossLiteAPI import BossLiteAPI
from ProdCommon.BossLite.Common.Exceptions import DbError
from ProdCommon.BossLite.Common.Exceptions import TaskError

from ProdCommon.BossLite.DbObjects.Job import Job
from ProdCommon.BossLite.DbObjects.Task import Task
from ProdCommon.BossLite.DbObjects.RunningJob import RunningJob


class DBinterface:
    def __init__(self, cfg_params):

        self.cfg_params = cfg_params

        self.db_type =  cfg_params.get("USER.use_db",'SQLite')
        return


    def configureDB(self):

        dbname = common.work_space.shareDir()+'crabDB'
        dbConfig = {'dbName':dbname
            }
        try: 
            common.bossSession = BossLiteAPI( self.db_type, dbConfig)
        except Exception, e :
            raise CrabException('Istantiate DB Session : '+str(e))

        try:
            common.bossSession.bossLiteDB.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql') 
        except Exception, e :
            raise CrabException('DB Installation error : '+str(e))
        return 
 
    def loadDB(self):

        dbname = common.work_space.shareDir()+'crabDB'
        dbConfig = {'dbName':dbname
            }
        try:
            common.bossSession = BossLiteAPI( self.db_type, dbConfig)
        except Exception, e :
            raise CrabException('Istantiate DB Session : '+str(e))

        return
 
    def getTask(self, jobsList='all'): 
        """
        Return task with all/list of jobs 
        """
        try:
            task = common.bossSession.load(1,jobsList)[0]
        except Exception, e :
            common.logger.debug(3, "Error while getting task : " +str(traceback.format_exc()))
            raise CrabException('Error while getting task '+str(e))
        return task

    def getJob(self, n): 
        """
        Return a task with a single job 
        """ 
        try:
            task = common.bossSession.load(1,str(n))[0]
        except Exception, e :
            common.logger.debug(3, "Error while getting job : " +str(traceback.format_exc()))
            raise CrabException('Error while getting job '+str(e))
        return task


    def createTask_(self, optsToSave):       
        """
        Task declaration
        with the first coniguration stuff 
        """
        opt={}
        if optsToSave.get('server_mode',0) == 1: opt['serverName']=optsToSave['server_name'] 
        opt['name']=common.work_space.taskName()  
        task = Task( opt )
        try:
            common.bossSession.saveTask( task )
        except Exception, e :
           # common.logger.debug(3, "Error creating task : " +str(traceback.format_exc()))
           # raise CrabException('Error creating task '+str(e))
            raise CrabException('Error creating task '+str(traceback.format_exc()))
            
        return 

    def updateTask_(self,optsToSave):       
        """
        Update task fields   
        """
        task = self.getTask()
   
        for key in optsToSave.keys():
            task[key] = optsToSave[key]
        try:
            common.bossSession.updateDB( task )
        except Exception, e :
            raise CrabException('Error updating task '+str(traceback.format_exc()))

        return 

    def createJobs_(self, jobsL, isNew=True):
        """  
        Fill crab DB with  the jobs filed 
        """
        task = self.getTask()

        jobs = [] 
        for id in jobsL:
            parameters = {}
            parameters['jobId'] = int(id)
            parameters['taskId'] = 1
            parameters['name'] = task['name'] + '_' + 'job' + str(id)
            job = Job(parameters)
            jobs.append(job) 
            common.bossSession.getRunningInstance(job)
            job.runningJob['status'] = 'C'
        ## added to support second step creation
        ## maybe it is not needed. TO CLARIFY
        if isNew:
            task.addJobs(jobs)
        else:
            task.appendJobs(jobs)
        try:
            common.bossSession.updateDB( task )
        except Exception, e :
            raise CrabException('Error updating task '+str(traceback.format_exc()))

        return

    def updateJob_(self, jobsL, optsToSave):       
        """
        Update Job fields   
        """
        task = self.getTask(jobsL)
        id =0 
        for job in task.jobs:
            for key in optsToSave[id].keys():
                job[key] = optsToSave[id][key]
            id+=1
        try:
            common.bossSession.updateDB( task )
        except Exception, e :
            raise CrabException('Error updating task '+str(traceback.format_exc()))
        return 

    def updateRunJob_(self, jobsL, optsToSave):       
        """
        Update Running Job fields   
        """
        task = self.getTask(jobsL)

        id=0
        for job in task.jobs:
            common.bossSession.getRunningInstance(job)
            for key in optsToSave[id].keys():
                job.runningJob[key] = optsToSave[id][key]
            id+=1
        common.bossSession.updateDB( task )
        return 

    def nJobs(self,list=''):
        
        task = self.getTask()
        listId=[]
        if list == 'list':
            for job in task.jobs:listId.append(int(job['jobId']))  
            return listId
        else:
            return len(task.jobs) 

    def dump(self,jobs):
        """
         List a complete set of infos for a job/range of jobs   
        """
        task = self.getTask(jobs)

        print "--------------------------"
        for Job in task.jobs:
            print "Id: ",Job['jobId']
            print "Dest: ", Job['dlsDestination']
            print "Output: ", Job['outputFiles']
            print "Args: ",Job['arguments']
            print "Service: ",Job.runningJob['service']
            print "--------------------------"
        return      

    def serializeTask(self, tmp_task = None):
        if tmp_task is None:
            tmp_task = self.getTask()
        return common.bossSession.serialize(tmp_task)   
 
    def queryID(self,server_mode=0):
        '''
        Return the taskId if serevr_mode =1 
        Return the joblistId if serevr_mode =0 
        '''     
        header=''
        lines=[]
        task = self.getTask()
        if server_mode == 1:
            header= "Task Id = %-40s " %(task['name'])
        else:
            for job in task.jobs: 
                toPrint=''
                common.bossSession.getRunningInstance(job)
                toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
                lines.append(toPrint)
            header+= "%-5s %-50s " % ('Job:','ID' ) 
        displayReport(self,header,lines)
        return   

    def queryTask(self,attr):
        '''
        Perform a query over a generic task attribute
        '''
        task = self.getTask()
        return task[attr]

    def queryJob(self, attr, jobsL):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        task = self.getTask(jobsL)
        for job in task.jobs:
            lines.append(job[attr])
        return lines

    def queryRunJob(self, attr, jobsL):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        task = self.getTask(jobsL)
        for job in task.jobs:
            common.bossSession.getRunningInstance(job)
            lines.append(job.runningJob[attr])
        return lines

    def queryDistJob(self, attr):
        '''
        Returns the list of distinct value for a given job attributes 
        '''
        distAttr=[]
        try:
            task = common.bossSession.loadJobDist( 1, attr ) 
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By distinct Attr '+str(e))

        for i in task: distAttr.append(i[attr])   
        return  distAttr

    def queryDistJob_Attr(self, attr_1, attr_2, list):
        '''
        Returns the list of distinct value for a given job attribute 
        '''
        distAttr=[]
        try:
            task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list ) 
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By distinct Attr '+str(e))

        for i in task: distAttr.append(i[attr_1])  
        return  distAttr

    def queryAttrJob(self, attr, field):
        '''
        Returns the list of jobs matching the given attribute
        '''
        matched=[]
        try:
            task = common.bossSession.loadJobsByAttr(attr ) 
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By Attr '+str(e))
        for i in task:
            matched.append(i[field])
        return  matched


    def queryAttrRunJob(self, attr,field):
        '''
        Returns the list of jobs matching the given attribute
        '''
        matched=[]
        try:
            task = common.bossSession.loadJobsByRunningAttr(attr)
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By Running Attr '+str(e))
        for i in task:
            matched.append(i.runningJob[field])
        return matched 

    def newRunJobs(self,nj='all'):
        """
        Get new running instances
        """  
        task = self.getTask(nj)

        for job in task.jobs:
            common.bossSession.getNewRunningInstance(job)
            job.runningJob['status'] = 'C'
            job.runningJob['statusScheduler'] = 'Created'
        common.bossSession.updateDB(task)     
        return        

    def deserXmlStatus(self, reportList):

        task = self.getTask()
        for job in task.jobs:
            if not job.runningJob:
                raise CrabException( "Missing running object for job %s"%str(job['jobId']) )

            id = str(job.runningJob['jobId'])
            rForJ = None
            nj_list= []
            for r in reportList:
                if r.getAttribute('id') in [ id, 'all']:
                    rForJ = r
                    break
            ## Check the submission number and create new running jobs on the client side          
            if int(job.runningJob['submission']) < int(rForJ.getAttribute('resubmit')) + 1:
                nj_list.append(id)
                self.newRunJobs(nj_list)

        task_new = self.getTask()

        for job in task_new.jobs:
            id = str(job.runningJob['jobId'])
            # TODO linear search, probably it can be optized with binary search
            rForJ = None
            for r in reportList:
                if r.getAttribute('id') in [ id, 'all']:
                    rForJ = r
                    break  
            # Data alignment
            jobStatus = str(job.runningJob['statusScheduler'])
            if str(job.runningJob['status']) != 'EE':
                if rForJ.getAttribute('statusScheduler') not in ['Created', 'Submitting', 'Unknown'] and \
                         job.runningJob['statusScheduler'] != 'Cleared':
                    job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
                    jobStatus = str(job.runningJob['statusScheduler'])
                    job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
          
                job.runningJob['destination'] = str( rForJ.getAttribute('site') )
                dest = str(job.runningJob['destination']).split(':')[0]
          
                job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
                exe_exit_code = str(job.runningJob['applicationReturnCode'])
          
                job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
                job_exit_code = str(job.runningJob['wrapperReturnCode'])
          
                #if str( rForJ.getAttribute('resubmit') ).isdigit():
                #    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
                #    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
          
                # TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio

        common.bossSession.updateDB( task_new )
        return

    # FIXME temporary method to verify what kind of submission to perform towards the server
    def checkIfNeverSubmittedBefore(self):
        for j in self.getTask().jobs:
            if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C' or \
                    j.runningJob['statusScheduler'] != 'Created':
                return False
        return True


Revision:	1.35
Committed:	Thu Jun 12 18:43:58 2008 UTC (16 years, 10 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_3_0, CRAB_2_3_0_pre6, CRAB_2_3_0_pre1
Branch point for:	CRAB_2_3_0_br
Changes since 1.34:	+21 -20 lines
Log Message:	Added TEMPORARY check on retrieved output avoiding multiple getOutput
#	Content
1	from crab_logger import Logger
2	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	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
14	from ProdCommon.BossLite.DbObjects.RunningJob import RunningJob
15
16
17	class DBinterface:
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	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)[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
81	"""
82	opt={}
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	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 = 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, jobsL, isNew=True):
111	"""
112	Fill crab DB with the jobs filed
113	"""
114	task = self.getTask()
115
116	jobs = []
117	for id in jobsL:
118	parameters = {}
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	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, jobsL, optsToSave):
140	"""
141	Update Job fields
142	"""
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, jobsL, optsToSave):
156	"""
157	Update Running Job fields
158	"""
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,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 = self.getTask(jobs)
185
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):
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 " %(task['name'])
211	else:
212	for job in task.jobs:
213	toPrint=''
214	common.bossSession.getRunningInstance(job)
215	toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
216	lines.append(toPrint)
217	header+= "%-5s %-50s " % ('Job:','ID' )
218	displayReport(self,header,lines)
219	return
220
221	def queryTask(self,attr):
222	'''
223	Perform a query over a generic task attribute
224	'''
225	task = self.getTask()
226	return task[attr]
227
228	def queryJob(self, attr, jobsL):
229	'''
230	Perform a query for a range/all/single job
231	over a generic job attribute
232	'''
233	lines=[]
234	task = self.getTask(jobsL)
235	for job in task.jobs:
236	lines.append(job[attr])
237	return lines
238
239	def queryRunJob(self, attr, jobsL):
240	'''
241	Perform a query for a range/all/single job
242	over a generic job attribute
243	'''
244	lines=[]
245	task = self.getTask(jobsL)
246	for job in task.jobs:
247	common.bossSession.getRunningInstance(job)
248	lines.append(job.runningJob[attr])
249	return lines
250
251	def queryDistJob(self, attr):
252	'''
253	Returns the list of distinct value for a given job attributes
254	'''
255	distAttr=[]
256	try:
257	task = common.bossSession.loadJobDist( 1, attr )
258	except Exception, e :
259	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
260	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
261
262	for i in task: distAttr.append(i[attr])
263	return distAttr
264
265	def queryDistJob_Attr(self, attr_1, attr_2, list):
266	'''
267	Returns the list of distinct value for a given job attribute
268	'''
269	distAttr=[]
270	try:
271	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
272	except Exception, e :
273	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
274	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
275
276	for i in task: distAttr.append(i[attr_1])
277	return distAttr
278
279	def queryAttrJob(self, attr, field):
280	'''
281	Returns the list of jobs matching the given attribute
282	'''
283	matched=[]
284	try:
285	task = common.bossSession.loadJobsByAttr(attr )
286	except Exception, e :
287	common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
288	raise CrabException('Error loading Jobs By Attr '+str(e))
289	for i in task:
290	matched.append(i[field])
291	return matched
292
293
294	def queryAttrRunJob(self, attr,field):
295	'''
296	Returns the list of jobs matching the given attribute
297	'''
298	matched=[]
299	try:
300	task = common.bossSession.loadJobsByRunningAttr(attr)
301	except Exception, e :
302	common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
303	raise CrabException('Error loading Jobs By Running Attr '+str(e))
304	for i in task:
305	matched.append(i.runningJob[field])
306	return matched
307
308	def newRunJobs(self,nj='all'):
309	"""
310	Get new running instances
311	"""
312	task = self.getTask(nj)
313
314	for job in task.jobs:
315	common.bossSession.getNewRunningInstance(job)
316	job.runningJob['status'] = 'C'
317	job.runningJob['statusScheduler'] = 'Created'
318	common.bossSession.updateDB(task)
319	return
320
321	def deserXmlStatus(self, reportList):
322
323	task = self.getTask()
324	for job in task.jobs:
325	if not job.runningJob:
326	raise CrabException( "Missing running object for job %s"%str(job['jobId']) )
327
328	id = str(job.runningJob['jobId'])
329	rForJ = None
330	nj_list= []
331	for r in reportList:
332	if r.getAttribute('id') in [ id, 'all']:
333	rForJ = r
334	break
335	## Check the submission number and create new running jobs on the client side
336	if int(job.runningJob['submission']) < int(rForJ.getAttribute('resubmit')) + 1:
337	nj_list.append(id)
338	self.newRunJobs(nj_list)
339
340	task_new = self.getTask()
341
342	for job in task_new.jobs:
343	id = str(job.runningJob['jobId'])
344	# TODO linear search, probably it can be optized with binary search
345	rForJ = None
346	for r in reportList:
347	if r.getAttribute('id') in [ id, 'all']:
348	rForJ = r
349	break
350	# Data alignment
351	jobStatus = str(job.runningJob['statusScheduler'])
352	if str(job.runningJob['status']) != 'EE':
353	if rForJ.getAttribute('statusScheduler') not in ['Created', 'Submitting', 'Unknown'] and \
354	job.runningJob['statusScheduler'] != 'Cleared':
355	job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
356	jobStatus = str(job.runningJob['statusScheduler'])
357	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
358
359	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
360	dest = str(job.runningJob['destination']).split(':')[0]
361
362	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
363	exe_exit_code = str(job.runningJob['applicationReturnCode'])
364
365	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
366	job_exit_code = str(job.runningJob['wrapperReturnCode'])
367
368	#if str( rForJ.getAttribute('resubmit') ).isdigit():
369	# job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
370	# job.runningJob['submission'] = int(rForJ.getAttribute('resubmit'))
371
372	# TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
373
374	common.bossSession.updateDB( task_new )
375	return
376
377	# FIXME temporary method to verify what kind of submission to perform towards the server
378	def checkIfNeverSubmittedBefore(self):
379	for j in self.getTask().jobs:
380	if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C' or \
381	j.runningJob['statusScheduler'] != 'Created':
382	return False
383	return True
384
385