ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/DBinterface.py
(Generate patch)

Comparing COMP/CRAB/python/DBinterface.py (file contents):
Revision 1.2 by spiga, Tue Mar 4 15:50:44 2008 UTC vs.
Revision 1.26 by gcodispo, Tue May 20 16:47:51 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.bossLiteDB.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 <        
49 <        return
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
50         {'server_name': 'crabas.lnl.infn.it/data1/cms/', '-scheduler': 'glite', '-jobtype': 'cmssw', '-server_mode': '0'}
51
81          """
82          opt={}
83 <        if optsToSave['server_mode'] == 1: opt['serverName']=optsToSave['server_name']
84 <        opt['jobType']=optsToSave['jobtype']  
56 <        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)
69 <        
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)
126 <        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
127 <
128 <        njobs = len(jobs)
129 <        lines=[]
130 <        header=''
131 <     #   ##query the DB asking the right infos for runningJobs  TODO  DS
132 <     #   for job in jobs:
133 <     #       ## here the query over runngJobs  
134 <     #       pass
178 >        task = self.getTask(jobs)
179  
180 <
181 <     #   ##Define Header to show and Pass the query results,
182 <     #   ##  header and format to displayReport()   TODO  DS
183 <     #   if njobs == 1: plural = ''
184 <     #   else:          plural = 's'
185 <     #   header += 'Listing %d job%s:\n' % (njobs, plural)
186 <     #   header += ' :\n' % (---) ## TODO DS
187 <
144 <     #   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 151 | 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:
183 <            jobs = common.bossSession.loadJob(task['id'],i+1)
184 <            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])
197 <            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):
246 +        '''
247 +        Returns the list of distinct value for a given job attributes
248 +        '''
249 +        distAttr=[]
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 queryDistJob_Attr(self, attr_1, attr_2, list):
260 +        '''
261 +        Returns the list of distinct value for a given job attribute
262 +        '''
263 +        distAttr=[]
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 +        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['jobId']) )
322 +
323 +            id = str(job.runningJob['jobId'])
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