Source Code for Module workCmdlineBackEnd

  1  ''' 
  2      A class that implements the standard Qube! cmdrange jobtype in python. 
  3      It should be simpler to maintain and extend than the older C++ variant. 
  4  ''' 
  5  #====================================== 
  6  #  $Revision: #3 $ 
  7  #  $Change: 16052 $ 
  8  #====================================== 
  9   
 10   
 11  import os.path 
 12  import sys 
 13  import re 
 14  import time  
 15  import select 
 16  import pprint 
 17  import logging 
 18   
 19  import qb.backend.commandBackEnd 
 20  import qb.backend.logParser 
 21  import qb.backend.utils as backendUtils 
 22   
 23  try: 
 24      import subprocess 
 25  except ImportError: 
 26      if os.name == 'posix': 
 27          import popen2 
 28   
 29   
 30 -class WorkCmdlineBackEnd(qb.backend.commandBackEnd.CommandBackEnd): 
 31      ''' 
 32      A backend for a python-based cmdrange jobtype which allows for extensible log parsing. 
 33      ''' 
 34   
 35 -    def __init__(self, job): 
 36          super(WorkCmdlineBackEnd, self).__init__(job) 
 37   
 38          # submit the job with job['package']['dev'] = True to set the logging level to DEBUG 
 39          self.logging = logging.getLogger('%s' % self.__class__.__name__) 
 40   
 41 -    def executeWork(self): 
 42          ''' 
 43          Now that we've got a job from the supervisor and have fired up the python 
 44          interpreter in the child process, it's time to ask the supervisor for an  
 45          agenda item to process. 
 46           
 47          Everything we need to know to get the work done is in either the job's or 
 48          the work's package dict. 
 49          ''' 
 50          if self.dev: 
 51              pp = pprint.PrettyPrinter(indent=2, width=1) 
 52   
 53          while True: 
 54              work_status = 1 
 55   
 56              backendUtils.bannerPrint('Requesting work', fhList=[sys.stdout, sys.stderr]) 
 57              work = qb.requestwork() 
 58   
 59              if self.dev: 
 60                  print "WORK:" 
 61                  pp.pprint(work) 
 62               
 63              # Deal with the minimal set of work statuses 
 64              if work['status'] == 'complete': 
 65                  work_status = 0 
 66                  break 
 67              elif work['status'] == 'pending': 
 68                  # preempted -- bail out 
 69                  print 'job %s has been preempted' % self.job['id'] 
 70                  work_status = 0 
 71                  qb.reportjob('pending') 
 72                  break 
 73              elif work['status'] == 'blocked': 
 74                  # blocked -- perhaps part of a dependency chain 
 75                  print 'job %s has been blocked' % self.job['id'] 
 76                  work_status = 0 
 77                  qb.reportjob('blocked') 
 78                  break 
 79              elif work['status'] == 'waiting': 
 80                  # waiting -- rare, come back in QB_WAITING_TIMEOUT seconds 
 81                  print 'job %s will be back in %s seconds' % (self.job['id'], self.QB_WAITING_TIMEOUT) 
 82                  time.sleep(self.QB_WAITING_TIMEOUT) 
 83                  continue 
 84   
 85              if not work['package']: 
 86                  work['package'] = {} 
 87   
 88              #============================================================ 
 89              #   Do the actual work  
 90              #============================================================ 
 91              backendUtils.bannerPrint('Processing work: %s' % work['name']) 
 92   
 93              cmd = work['package']['cmdline'] 
 94              retCode = self.runCmd(work, cmd) 
 95   
 96              if retCode > 0: 
 97                  backendUtils.flushPrint('ERROR: Processing work failed\n', fhList=[sys.stdout, sys.stderr]) 
 98               
 99              work_status = retCode 
100   
101              if work.get('resultpackage') is None: 
102                  work['resultpackage'] = {} 
103              # ----------------------------------------------------------- 
104              # set the work status, then report it back to the supervisor 
105              # so that it can update the server-side agenda 
106              # ----------------------------------------------------------- 
107              if work_status != 0: 
108                  # either the work or the job instance itself has failed 
109                  work['status'] = 'failed' 
110              elif self.outputPaths_required and len(work.get('resultpackage', {}).get('outputPaths', '')) == 0: 
111                  work['status'] = 'failed' 
112                  backendUtils.flushPrint('WARNING: no "regex_outputPaths" match was found, setting agenda item status to "failed".', fhList=[sys.stdout, sys.stderr]) 
113              else: 
114                  # mark the work as complete, and reset both the failure counter and timer 
115                  work['status'] = 'complete' 
116   
117              # report the work status back to the supervisor 
118              backendUtils.bannerPrint('Reporting work as %(status)s: %(name)s ' % work, fhList=[sys.stdout, sys.stderr]) 
119   
120              qb.reportwork(work) 
121