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.7 by spiga, Mon Mar 17 14:00:59 2008 UTC vs.
Revision 1.35.4.2 by spiga, Mon Jul 7 13:24:31 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 <        self.task = common.bossSession.loadTaskByID(1)
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):
55 <
56 <        #if jobsList == 'all':
57 <        #    self.task = common.bossSession.loadTaskByID(1)
58 <        #else:
59 <        #self.task = common.bossSession.load('1','5')  
60 <        #return self.task[0]
61 <        self.task = common.bossSession.loadTaskByID(1)
62 <        return self.task
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 <        self.job = common.bossSession.loadJobByID(1,n)
68 <        return self.job
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
64        {'server_name': 'crabas.lnl.infn.it/data1/cms/', '-scheduler': 'glite', '-jobtype': 'cmssw', '-server_mode': '0'}
65
81          """
82          opt={}
83 <        if optsToSave['server_mode'] == 1: 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.loadTaskByName(common.work_space.taskName() )
100 <        task = common.bossSession.loadTaskByID(1)
81 <        
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, isNew=True):
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'] = 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, nj, optsToSave):      
139 >    def updateJob_(self, jobsL, optsToSave):      
140          """
141          Update Job fields  
142          """
143 <        task = common.bossSession.loadTaskByID(1)
144 <        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
145 <        for i in range(len(nj)):
146 <           # jobs = common.bossSession.loadJob(task['id'],i)
147 <            for key in optsToSave[i].keys():
148 <                task.jobs[i][key] = optsToSave[i][key]
149 <        common.bossSession.updateDB( task )
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, nj, optsToSave):      
155 >    def updateRunJob_(self, jobsL, optsToSave):      
156          """
157          Update Running Job fields  
158          """
159 <        task = common.bossSession.loadTaskByID(1)
160 <        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
161 <        for i in nj:
162 <            common.bossSession.getRunningInstance(task.jobs[i])
163 <            for key in optsToSave.keys():
164 <                task.jobs[i].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):
170 >    def nJobs(self,list=''):
171          
172 <        task = common.bossSession.loadTaskByID(1)
173 <        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
174 <        return len(task.jobs)
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 = common.bossSession.loadTaskByID(1)
140 <        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
141 <
142 <        njobs = len(jobs)
143 <        lines=[]
144 <        header=''
145 <     #   ##query the DB asking the right infos for runningJobs  TODO  DS
146 <     #   for job in jobs:
147 <     #       ## here the query over runngJobs  
148 <     #       pass
149 <
150 <
151 <     #   ##Define Header to show and Pass the query results,
152 <     #   ##  header and format to displayReport()   TODO  DS
153 <     #   if njobs == 1: plural = ''
154 <     #   else:          plural = 's'
155 <     #   header += 'Listing %d job%s:\n' % (njobs, plural)
156 <     #   header += ' :\n' % (---) ## TODO DS
184 >        task = self.getTask(jobs)
185  
186 <     #   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 = self.getTask()
199 +        return common.bossSession.serialize(tmp_task)  
200  
201      def queryID(self,server_mode=0):
202          '''
# Line 165 | Line 205 | class DBinterface:
205          '''    
206          header=''
207          lines=[]
208 <        task = common.bossSession.loadTaskByID(1)
208 >        task = self.getTask()
209          if server_mode == 1:
210              header= "Task Id = %-40s " %(task['name'])
211          else:
212 <         #   task = common.bossSession.loadTaskByName(common.work_space.taskName() )
213 <            for i in range(len(task.job)):
214 <                common.bossSession.getRunningInstance(task.jobs[i])
215 <                lines.append(task.jobs[i].runningJob['schedulerId'])
216 <          
217 <            header+= "Job: %-5s Id = %-40s: \n"
218 <        displayReport(header,lines)
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 = common.bossSession.loadTaskByID(1)
225 >        task = self.getTask()
226          return task[attr]
227  
228 <    def queryJob(self, attr, njobs):
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 = common.bossSession.loadTaskByID(1)
235 <        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
236 <        for i in njobs:
197 <            jobs = common.bossSession.loadJob(task['id'],i+1)
198 <            lines.append(task.jobs[i][attr])
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, jobs):
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 = common.bossSession.loadTaskByID(1)
246 <       # task = common.bossSession.loadTaskByName( common.work_space.taskName() )
247 <        for i in jobs:
248 <            common.bossSession.getRunningInstance(task.jobs[i-1])
211 <            lines.append(task.jobs[i-1].runningJob[attr])
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):
# Line 216 | Line 253 | class DBinterface:
253          Returns the list of distinct value for a given job attributes
254          '''
255          distAttr=[]
256 <        task = common.bossSession.loadJobDist( 1, attr )
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 attributes
267 >        Returns the list of distinct value for a given job attribute
268          '''
269          distAttr=[]
270 <        task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
271 <        for i in task: distAttr.append(i[attr_1])  
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):
# Line 234 | Line 281 | class DBinterface:
281          Returns the list of jobs matching the given attribute
282          '''
283          matched=[]
284 <        task = common.bossSession.loadJobsByAttr(attr )
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
# Line 245 | Line 296 | class DBinterface:
296          Returns the list of jobs matching the given attribute
297          '''
298          matched=[]
299 <        task = common.bossSession.loadJobsByRunningAttr(attr)
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[field])
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 +
351 +            # Data alignment
352 +            if rForJ.getAttribute('status') not in ['Created', 'Unknown'] and not\
353 +                (job.runningJob['statusScheduler'] == 'Killing' and rForJ.getAttribute('status')!='Killed') \
354 +                and not  (job.runningJob['statusScheduler'] in 'Submitting'  and rForJ.getAttribute('status') in ['Killed','Aborted','Cleared'])  :
355 +                   # update the status  
356 +                common.logger.debug(3,"Updating DB status for job: " + str(id) + " @: " \
357 +                                      + str(rForJ.getAttribute('status')) )
358 +                job.runningJob['statusScheduler'] = str( rForJ.getAttribute('status') )
359 +                job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
360 +          
361 +                job.runningJob['destination'] = str( rForJ.getAttribute('site') )
362 +                dest = str(job.runningJob['destination']).split(':')[0]
363 +              
364 +                job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
365 +                exe_exit_code = str(job.runningJob['applicationReturnCode'])
366 +              
367 +                job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
368 +                job_exit_code = str(job.runningJob['wrapperReturnCode'])
369 +          
370 +            #if str( rForJ.getAttribute('resubmit') ).isdigit():
371 +            #    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
372 +            #    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
373 +          
374 +            # TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
375 +
376 +        common.bossSession.updateDB( task_new )
377 +        return
378 +
379 +    # FIXME temporary method to verify what kind of submission to perform towards the server
380 +    def checkIfNeverSubmittedBefore(self):
381 +        for j in self.getTask().jobs:
382 +            if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C' or \
383 +                    j.runningJob['statusScheduler'] != 'Created':
384 +                return False
385 +        return True
386 +
387 +

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines