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.15 by mcinquil, Tue Apr 15 14:13:07 2008 UTC vs.
Revision 1.48 by spiga, Mon Nov 10 13:30:23 2008 UTC

# Line 34 | Line 34 | class DBinterface:
34              raise CrabException('Istantiate DB Session : '+str(e))
35  
36          try:
37 <            common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')    
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
# Line 56 | Line 56 | class DBinterface:
56          Return task with all/list of jobs
57          """
58          try:
59 <            task = common.bossSession.load(1,jobsList)[0]
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))
# Line 67 | Line 67 | class DBinterface:
67          Return a task with a single job
68          """
69          try:
70 <            task = common.bossSession.load(1,str(n))[0]
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))
# Line 107 | Line 107 | class DBinterface:
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          """
# Line 116 | Line 116 | class DBinterface:
116          jobs = []
117          for id in jobsL:
118              parameters = {}
119 <            parameters['jobId'] =  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)
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 :
# Line 173 | Line 181 | class DBinterface:
181          """
182           List a complete set of infos for a job/range of jobs  
183          """
184 <        task = self.getTask()
177 <
178 <        njobs = len(jobs)
179 <        lines=[]
180 <        header=''
181 <     #   ##query the DB asking the right infos for runningJobs  TODO  DS
182 <     #   for job in jobs:
183 <     #       ## here the query over runngJobs  
184 <     #       pass
185 <
186 <
187 <     #   ##Define Header to show and Pass the query results,
188 <     #   ##  header and format to displayReport()   TODO  DS
189 <     #   if njobs == 1: plural = ''
190 <     #   else:          plural = 's'
191 <     #   header += 'Listing %d job%s:\n' % (njobs, plural)
192 <     #   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):
# Line 199 | Line 198 | class DBinterface:
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
# Line 208 | Line 207 | class DBinterface:
207          lines=[]
208          task = self.getTask()
209          if server_mode == 1:
210 <            header= "Task Id = %-40s " %(task['name'])
211 <        else:
210 >            headerTask= "Task Id = %-40s " %(task['name'])
211 >            displayReport(self,headerTask,lines)
212 >        if (jid ) or (server_mode == 0):
213              for job in task.jobs:
214                  toPrint=''
215                  common.bossSession.getRunningInstance(job)
216 <                toPrint = "%-5s %-50s " % (job['id'],job.runningJob['schedulerId'])
216 >                toPrint = "%-5s %-50s " % (job['jobId'],job.runningJob['schedulerId'])
217                  lines.append(toPrint)
218              header+= "%-5s %-50s " % ('Job:','ID' )
219 <        displayReport(self,header,lines)
219 >            displayReport(self,header,lines)
220          return  
221  
222      def queryTask(self,attr):
# Line 234 | Line 234 | class DBinterface:
234          lines=[]
235          task = self.getTask(jobsL)
236          for job in task.jobs:
237 <            lines.append(eval(job[attr]))
237 >            lines.append(job[attr])
238          return lines
239  
240      def queryRunJob(self, attr, jobsL):
# Line 260 | Line 260 | class DBinterface:
260              common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
261              raise CrabException('Error loading Jobs By distinct Attr '+str(e))
262  
263 <        for i in task: distAttr.append(eval(i[attr]))  
263 >        for i in task: distAttr.append(i[attr])  
264          return  distAttr
265  
266      def queryDistJob_Attr(self, attr_1, attr_2, list):
# Line 274 | Line 274 | class DBinterface:
274              common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
275              raise CrabException('Error loading Jobs By distinct Attr '+str(e))
276  
277 <        for i in task: distAttr.append(eval(i[attr_1]))  
277 >        for i in task: distAttr.append(i[attr_1])  
278          return  distAttr
279  
280      def queryAttrJob(self, attr, field):
# Line 303 | Line 303 | class DBinterface:
303              common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
304              raise CrabException('Error loading Jobs By Running Attr '+str(e))
305          for i in task:
306 <            matched.append(i[field])
306 >            matched.append(i.runningJob[field])
307          return matched
308 +
309 +    def newRunJobs(self,nj='all'):
310 +        """
311 +        Get new running instances
312 +        """  
313 +        task = self.getTask(nj)
314 +
315 +        for job in task.jobs:
316 +            common.bossSession.getNewRunningInstance(job)
317 +            job.runningJob['status'] = 'C'
318 +            job.runningJob['statusScheduler'] = 'Created'
319 +        common.bossSession.updateDB(task)    
320 +        return        
321 +
322 +    def deserXmlStatus(self, reportList):
323 +
324 +        task = self.getTask()
325 +        for job in task.jobs:
326 +            if not job.runningJob:
327 +                raise CrabException( "Missing running object for job %s"%str(job['jobId']) )
328 +
329 +            id = str(job.runningJob['jobId'])
330 +            rForJ = None
331 +            nj_list= []
332 +            for r in reportList:
333 +                if r.getAttribute('id') in [ id, 'all']:
334 +                    rForJ = r
335 +                    break
336 +            ## Check the submission number and create new running jobs on the client side          
337 +            if rForJ.getAttribute('resubmit') != 'None' and (rForJ.getAttribute('status') not in ['Killed','Done','Done (Failed)']) :
338 +                if int(job.runningJob['submission']) < int(rForJ.getAttribute('resubmit')) + 1:
339 +                    nj_list.append(id)
340 +            if len(nj_list) > 0: self.newRunJobs(nj_list)
341 +
342 +        task_new = self.getTask()
343 +
344 +        for job in task_new.jobs:
345 +            id = str(job.runningJob['jobId'])
346 +            # TODO linear search, probably it can be optized with binary search
347 +            rForJ = None
348 +            for r in reportList:
349 +                if r.getAttribute('id') in [ id, 'all']:
350 +                    rForJ = r
351 +                    break
352 +                  
353 +            # Data alignment
354 +            if rForJ.getAttribute('status') not in ['Created', 'Unknown']:
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 +                if (rForJ.getAttribute('status') == 'Done' or rForJ.getAttribute('status') == 'Done (Failed)')\
360 +                  and rForJ.getAttribute('sched_status') == 'E' :
361 +                    job.runningJob['status'] = 'SD'
362 +                else:
363 +                    job.runningJob['status'] = str( rForJ.getAttribute('sched_status') )
364 +          
365 +                job.runningJob['schedulerId'] = str( rForJ.getAttribute('sched_id') )
366 +
367 +                job.runningJob['destination'] = str( rForJ.getAttribute('site') )
368 +                dest = str(job.runningJob['destination']).split(':')[0]
369 +              
370 +                job.runningJob['applicationReturnCode'] = str( rForJ.getAttribute('exe_exit') )
371 +                exe_exit_code = str(job.runningJob['applicationReturnCode'])
372 +              
373 +                job.runningJob['wrapperReturnCode'] = str( rForJ.getAttribute('job_exit') )
374 +                job_exit_code = str(job.runningJob['wrapperReturnCode'])
375 +          
376 +            #if str( rForJ.getAttribute('resubmit') ).isdigit():
377 +            #    job['submissionNumber'] = int(rForJ.getAttribute('resubmit'))
378 +            #    job.runningJob['submission'] =  int(rForJ.getAttribute('resubmit'))
379 +          
380 +            # TODO cleared='0' field, how should it be handled/mapped in BL? #Fabio
381 +
382 +        common.bossSession.updateDB( task_new )
383 +        return
384 +
385 +    # FIXME temporary method to verify what kind of submission to perform towards the server
386 +    def checkIfNeverSubmittedBefore(self):
387 +        for j in self.getTask().jobs:
388 +            if j.runningJob['submission'] > 1 or j.runningJob['status'] != 'C' or \
389 +                    j.runningJob['statusScheduler'] != 'Created':
390 +                return False
391 +        return True
392 +
393 +

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines