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.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):
        """  
        Fill crab DB with  the jobs filed 
        """
        task = self.getTask()

        jobs = [] 
        for id in jobsL:
            parameters = {}
            parameters['jobId'] =  str(id)
            parameters['name'] = task['name'] + '_' + 'job' + str(id)
            job = Job(parameters)
            jobs.append(job)  
        task.addJobs(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)

        Jobs = task.getJobs()
        print "--------------------------"
        for Job in Jobs:
            print "Id: ",Job['id']
            print "Dest: ", Job['dlsDestination']
            print "Output: ", Job['outputFiles']
            print "Args: ",Job['arguments']
            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['id'],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(eval(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(eval(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(eval(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 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['id']) )

            id = str(job.runningJob['id'])
            # 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 rForJ.getAttribute('status') not in ['Created', 'Submitting']:
                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'))
            # TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio

        common.bossSession.updateDB( task )

        return
Revision:	1.18
Committed:	Sun Apr 20 17:04:52 2008 UTC (17 years ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.17:	+1 -1 lines
Log Message:	bug fix
#	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.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):
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'] = str(id)
120	parameters['name'] = task['name'] + '_' + 'job' + str(id)
121	job = Job(parameters)
122	jobs.append(job)
123	task.addJobs(jobs)
124	try:
125	common.bossSession.updateDB( task )
126	except Exception, e :
127	raise CrabException('Error updating task '+str(traceback.format_exc()))
128
129	return
130
131	def updateJob_(self, jobsL, optsToSave):
132	"""
133	Update Job fields
134	"""
135	task = self.getTask(jobsL)
136	id =0
137	for job in task.jobs:
138	for key in optsToSave[id].keys():
139	job[key] = optsToSave[id][key]
140	id+=1
141	try:
142	common.bossSession.updateDB( task )
143	except Exception, e :
144	raise CrabException('Error updating task '+str(traceback.format_exc()))
145	return
146
147	def updateRunJob_(self, jobsL, optsToSave):
148	"""
149	Update Running Job fields
150	"""
151	task = self.getTask(jobsL)
152
153	id=0
154	for job in task.jobs:
155	common.bossSession.getRunningInstance(job)
156	for key in optsToSave[id].keys():
157	job.runningJob[key] = optsToSave[id][key]
158	id+=1
159	common.bossSession.updateDB( task )
160	return
161
162	def nJobs(self,list=''):
163
164	task = self.getTask()
165	listId=[]
166	if list == 'list':
167	for job in task.jobs:listId.append(int(job['jobId']))
168	return listId
169	else:
170	return len(task.jobs)
171
172	def dump(self,jobs):
173	"""
174	List a complete set of infos for a job/range of jobs
175	"""
176	task = self.getTask(jobs)
177
178	Jobs = task.getJobs()
179	print "--------------------------"
180	for Job in Jobs:
181	print "Id: ",Job['id']
182	print "Dest: ", Job['dlsDestination']
183	print "Output: ", Job['outputFiles']
184	print "Args: ",Job['arguments']
185	print "--------------------------"
186	return
187
188	def serializeTask(self, tmp_task = None):
189	if tmp_task is None:
190	tmp_task = self.getTask()
191	return common.bossSession.serialize(tmp_task)
192
193	def queryID(self,server_mode=0):
194	'''
195	Return the taskId if serevr_mode =1
196	Return the joblistId if serevr_mode =0
197	'''
198	header=''
199	lines=[]
200	task = self.getTask()
201	if server_mode == 1:
202	header= "Task Id = %-40s " %(task['name'])
203	else:
204	for job in task.jobs:
205	toPrint=''
206	common.bossSession.getRunningInstance(job)
207	toPrint = "%-5s %-50s " % (job['id'],job.runningJob['schedulerId'])
208	lines.append(toPrint)
209	header+= "%-5s %-50s " % ('Job:','ID' )
210	displayReport(self,header,lines)
211	return
212
213	def queryTask(self,attr):
214	'''
215	Perform a query over a generic task attribute
216	'''
217	task = self.getTask()
218	return task[attr]
219
220	def queryJob(self, attr, jobsL):
221	'''
222	Perform a query for a range/all/single job
223	over a generic job attribute
224	'''
225	lines=[]
226	task = self.getTask(jobsL)
227	for job in task.jobs:
228	lines.append(eval(job[attr]))
229	return lines
230
231	def queryRunJob(self, attr, jobsL):
232	'''
233	Perform a query for a range/all/single job
234	over a generic job attribute
235	'''
236	lines=[]
237	task = self.getTask(jobsL)
238	for job in task.jobs:
239	common.bossSession.getRunningInstance(job)
240	lines.append(job.runningJob[attr])
241	return lines
242
243	def queryDistJob(self, attr):
244	'''
245	Returns the list of distinct value for a given job attributes
246	'''
247	distAttr=[]
248	try:
249	task = common.bossSession.loadJobDist( 1, attr )
250	except Exception, e :
251	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
252	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
253
254	for i in task: distAttr.append(eval(i[attr]))
255	return distAttr
256
257	def queryDistJob_Attr(self, attr_1, attr_2, list):
258	'''
259	Returns the list of distinct value for a given job attribute
260	'''
261	distAttr=[]
262	try:
263	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
264	except Exception, e :
265	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
266	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
267
268	for i in task: distAttr.append(eval(i[attr_1]))
269	return distAttr
270
271	def queryAttrJob(self, attr, field):
272	'''
273	Returns the list of jobs matching the given attribute
274	'''
275	matched=[]
276	try:
277	task = common.bossSession.loadJobsByAttr(attr )
278	except Exception, e :
279	common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
280	raise CrabException('Error loading Jobs By Attr '+str(e))
281	for i in task:
282	matched.append(i[field])
283	return matched
284
285
286	def queryAttrRunJob(self, attr,field):
287	'''
288	Returns the list of jobs matching the given attribute
289	'''
290	matched=[]
291	try:
292	task = common.bossSession.loadJobsByRunningAttr(attr)
293	except Exception, e :
294	common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
295	raise CrabException('Error loading Jobs By Running Attr '+str(e))
296	for i in task:
297	matched.append(i.runningJob[field])
298	return matched
299
300	def deserXmlStatus(self, reportList):
301
302	task = self.getTask()
303
304	for job in task.jobs:
305	if not job.runningJob:
306	raise CrabException( "Missing running object for job %s"%str(job['id']) )
307
308	id = str(job.runningJob['id'])
309	# TODO linear search, probably it can be optized with binary search
310	rForJ = None
311	for r in reportList:
312	if r.getAttribute('id') in [ id, 'all']:
313	rForJ = r
314	break
315
316	# Data alignment
317	jobStatus = str(job.runningJob['statusScheduler'])
318	if rForJ.getAttribute('status') not in ['Created', 'Submitting']:
319	job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
320	jobStatus = str(job.runningJob['statusScheduler'])
321	job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
322
323	job.runningJob['destination'] = str( rForJ.getAttribute('site') )
324	dest = str(job.runningJob['destination']).split(':')[0]
325
326	job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
327	exe_exit_code = str(job.runningJob['applicationReturnCode'])
328
329	job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
330	job_exit_code = str(job.runningJob['wrapperReturnCode'])
331
332	if str( rForJ.getAttribute('resubmit') ).isdigit():
333	job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
334	# TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
335
336	common.bossSession.updateDB( task )
337
338	return