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.9 by spiga, Fri Mar 21 16:28:07 2008 UTC vs.
Revision 1.53 by mcinquil, Thu Feb 5 15:44:02 2009 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 <        self.task = common.bossSession.load(1)[0]
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'): #, cfg_params):
54 >    def getTask(self, jobsList='all'):
55          """
56          Return task with all/list of jobs
57          """
58 <
59 <        if jobsList == 'all':
60 <            task = common.bossSession.load(1)[0]
61 <        else:
62 <            if len(jobsList)>1: str_jobs=string.join(map(str,jobsList),",")
53 <            else: str_jobs=str(jobsList)  
54 <            task = common.bossSession.load(1,jobsList)[0]
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 <        task = common.bossSession.load(1,str(n))[0]
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  
# Line 68 | Line 80 | class DBinterface:
80          with the first coniguration stuff
81          """
82          opt={}
83 <        opt['serverName']=optsToSave['server_name']
84 <        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 )
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.load(1)[0]
100 <        
99 >        task = self.getTask()
100 >  
101          for key in optsToSave.keys():
102              task[key] = optsToSave[key]
103 <        common.bossSession.updateDB( task )
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):
110 >    def createJobs_(self, jobsL, isNew=True):
111          """  
112          Fill crab DB with  the jobs filed
113          """
114 <        task = common.bossSession.loadTask(1)
114 >        task = self.getTask()
115 >
116          jobs = []
117          for id in jobsL:
118              parameters = {}
119 <            parameters['jobId'] =  str(id)
120 <            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, jobsL, optsToSave):      
140          """
141          Update Job fields  
142          """
143 <        task = common.bossSession.load(1)[0]
144 <        for id in jobsL:
143 >        task = self.getTask(jobsL)
144 >        id =0
145 >        for job in task.jobs:
146              for key in optsToSave[id].keys():
147 <                task.jobs[id][key] = optsToSave[id][key]
148 <        common.bossSession.updateDB( task )
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 = common.bossSession.load(1)[0]
160 <        for id in jobsL:
161 <            common.bossSession.getRunningInstance(task.jobs[id])
162 <            for key in optsToSave.keys():
163 <                task.jobs[id].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,list=''):
171          
172 <        task = common.bossSession.load(1)[0]
172 >        task = self.getTask()
173          listId=[]
174          if list == 'list':
175              for job in task.jobs:listId.append(int(job['jobId']))  
# Line 139 | Line 181 | class DBinterface:
181          """
182           List a complete set of infos for a job/range of jobs  
183          """
184 <        task = common.bossSession.load(1)[0]
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 <
152 <
153 <     #   ##Define Header to show and Pass the query results,
154 <     #   ##  header and format to displayReport()   TODO  DS
155 <     #   if njobs == 1: plural = ''
156 <     #   else:          plural = 's'
157 <     #   header += 'Listing %d job%s:\n' % (njobs, plural)
158 <     #   header += ' :\n' % (---) ## TODO DS
159 <
160 <     #   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 = common.bossSession.loadTaskByID(1)
166 <            tmp_task = common.bossSession.load(1)[0]
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 = common.bossSession.load(1)[0]
208 >        task = self.getTask()
209          if server_mode == 1:
210 <            header= "Task Id = %-40s " %(task['name'])
211 <        else:
212 <            for i in range(len(task.job)):
213 <                common.bossSession.getRunningInstance(task.jobs[i])
214 <                lines.append(task.jobs[i].runningJob['schedulerId'])
215 <          
216 <            header+= "Job: %-5s Id = %-40s: \n"
217 <        displayReport(header,lines)
210 >            # init client server params...
211 >            CliServerParams(self)      
212 >            headerTask= "Task Id = %-40s " %(task['name'])
213 >            displayReport(self,headerTask,lines)
214 >            common.logger.message(showWebMon(self.server_name))
215 >        if (jid ) or (server_mode == 0):
216 >            for job in task.jobs:
217 >                toPrint=''
218 >                common.bossSession.getRunningInstance(job)
219 >                toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
220 >                lines.append(toPrint)
221 >            header+= "%-5s %-50s " % ('Job:','ID' )
222 >            displayReport(self,header,lines)
223          return  
224  
225      def queryTask(self,attr):
226          '''
227          Perform a query over a generic task attribute
228          '''
229 <        task = common.bossSession.loadTask(1)
229 >        task = self.getTask()
230          return task[attr]
231  
232 <    def queryJob(self, attr, jobs):
232 >    def queryJob(self, attr, jobsL):
233          '''
234          Perform a query for a range/all/single job
235          over a generic job attribute
236          '''
237          lines=[]
238 <        str_jobs=string.join(map(str,jobs),",")
202 <        task = common.bossSession.load(1,str_jobs)[0]
238 >        task = self.getTask(jobsL)
239          for job in task.jobs:
240 <            lines.append(eval(job[attr]))
240 >            lines.append(job[attr])
241          return lines
242  
243 <    def queryRunJob(self, attr, jobs):
243 >    def queryRunJob(self, attr, jobsL):
244          '''
245          Perform a query for a range/all/single job
246          over a generic job attribute
247          '''
248          lines=[]
249 <        str_jobs=string.join(map(str,jobs),",")
214 <        task = common.bossSession.load(1,str_jobs)[0]
249 >        task = self.getTask(jobsL)
250          for job in task.jobs:
251              common.bossSession.getRunningInstance(job)
252              lines.append(job.runningJob[attr])
# Line 222 | Line 257 | class DBinterface:
257          Returns the list of distinct value for a given job attributes
258          '''
259          distAttr=[]
260 <        task = common.bossSession.loadJobDist( 1, attr )
261 <        for i in task: distAttr.append(eval(i[attr]))  
260 >        try:
261 >            task = common.bossSession.loadJobDist( 1, attr )
262 >        except Exception, e :
263 >            common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
264 >            raise CrabException('Error loading Jobs By distinct Attr '+str(e))
265 >
266 >        for i in task: distAttr.append(i[attr])  
267          return  distAttr
268  
269      def queryDistJob_Attr(self, attr_1, attr_2, list):
# Line 231 | Line 271 | class DBinterface:
271          Returns the list of distinct value for a given job attribute
272          '''
273          distAttr=[]
274 <        task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
275 <        for i in task: distAttr.append(eval(i[attr_1]))  
274 >        try:
275 >            task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
276 >        except Exception, e :
277 >            common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
278 >            raise CrabException('Error loading Jobs By distinct Attr '+str(e))
279 >
280 >        for i in task: distAttr.append(i[attr_1])  
281          return  distAttr
282  
283      def queryAttrJob(self, attr, field):
# Line 240 | Line 285 | class DBinterface:
285          Returns the list of jobs matching the given attribute
286          '''
287          matched=[]
288 <        task = common.bossSession.loadJobsByAttr(attr )
288 >        try:
289 >            task = common.bossSession.loadJobsByAttr(attr )
290 >        except Exception, e :
291 >            common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
292 >            raise CrabException('Error loading Jobs By Attr '+str(e))
293          for i in task:
294              matched.append(i[field])
295          return  matched
# Line 251 | Line 300 | class DBinterface:
300          Returns the list of jobs matching the given attribute
301          '''
302          matched=[]
303 <        task = common.bossSession.loadJobsByRunningAttr(attr)
303 >        try:
304 >            task = common.bossSession.loadJobsByRunningAttr(attr)
305 >        except Exception, e :
306 >            common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
307 >            raise CrabException('Error loading Jobs By Running Attr '+str(e))
308          for i in task:
309 <            matched.append(i[field])
309 >            matched.append(i.runningJob[field])
310          return matched
311 +
312 +    def newRunJobs(self,nj='all'):
313 +        """
314 +        Get new running instances
315 +        """  
316 +        task = self.getTask(nj)
317 +
318 +        for job in task.jobs:
319 +            common.bossSession.getNewRunningInstance(job)
320 +            job.runningJob['status'] = 'C'
321 +            job.runningJob['statusScheduler'] = 'Created'
322 +        common.bossSession.updateDB(task)    
323 +        return        
324 +
325 +    def deserXmlStatus(self, reportList):
326 +
327 +        task = self.getTask()
328 +        for job in task.jobs:
329 +            if not job.runningJob:
330 +                raise CrabException( "Missing running object for job %s"%str(job['jobId']) )
331 +
332 +            id = str(job.runningJob['jobId'])
333 +            rForJ = None
334 +            nj_list= []
335 +            for r in reportList:
336 +                if r.getAttribute('id') in [ id, 'all']:
337 +                    rForJ = r
338 +                    break
339 +            ## Check the submission number and create new running jobs on the client side
340 +            if rForJ.getAttribute('resubmit') != 'None' and (rForJ.getAttribute('status') not in ['Cleared','Killed','Done','Done (Failed)','Not Submitted']) :
341 +                if int(job.runningJob['submission']) < int(rForJ.getAttribute('resubmit')) + 1:
342 +                    nj_list.append(id)
343 +            if len(nj_list) > 0: self.newRunJobs(nj_list)
344 +
345 +        task_new = self.getTask()
346 +
347 +        for job in task_new.jobs:
348 +            id = str(job.runningJob['jobId'])
349 +            # TODO linear search, probably it can be optized with binary search
350 +            rForJ = None
351 +            for r in reportList:
352 +                if r.getAttribute('id') in [ id, 'all']:
353 +                    rForJ = r
354 +                    break
355 +                  
356 +            # Data alignment
357 +            if rForJ.getAttribute('status') not in ['Created', 'Unknown']:
358 +                   # update the status  
359 +                common.logger.debug(3,"Updating DB status for job: " + str(id) + " @: " \
360 +                                      + str(rForJ.getAttribute('status')) )
361 +                job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
362 +                if (rForJ.getAttribute('status') == 'Done' or rForJ.getAttribute('status') == 'Done (Failed)')\
363 +                  and rForJ.getAttribute('sched_status') == 'E' :
364 +                    job.runningJob['status'] = 'SD'
365 +                else:
366 +                    job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
367 +          
368 +                job.runningJob['schedulerId'] = str( rForJ.getAttribute('sched_id') )
369 +
370 +                job.runningJob['destination'] = str( rForJ.getAttribute('site') )
371 +                dest = str(job.runningJob['destination']).split(':')[0]
372 +              
373 +                job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
374 +                exe_exit_code = str(job.runningJob['applicationReturnCode'])
375 +              
376 +                job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
377 +                job_exit_code = str(job.runningJob['wrapperReturnCode'])
378 +
379 +                ## unsing 'standardInput' field for 'ended' tag ['Y','N']
380 +                job['standardInput'] = str( rForJ.getAttribute('ended') )
381 +          
382 +            #if str( rForJ.getAttribute('resubmit') ).isdigit():
383 +            #    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
384 +            #    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
385 +          
386 +            # TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
387 +
388 +        common.bossSession.updateDB( task_new )
389 +        return
390 +
391 +    # FIXME temporary method to verify what kind of submission to perform towards the server
392 +    def checkIfNeverSubmittedBefore(self):
393 +        for j in self.getTask().jobs:
394 +            if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C' or \
395 +                    j.runningJob['statusScheduler'] != 'Created':
396 +                return False
397 +        return True
398 +
399 +

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines