#!/usr/bin/env python #----------------------------------------------------- # This job wrapper script is automatically created by # GANGA LCG backend handler. # # It controls: # 1. unpack input sandbox # 2. invoke application executable # 3. invoke monitoring client #----------------------------------------------------- import os,os.path,shutil,tempfile import sys,popen2,time,traceback ## Utility functions ## def timeString(): return time.strftime('%a %b %d %H:%M:%S %Y',time.gmtime(time.time())) def printInfo(s): out.write(timeString() + ' [Info]' + ' ' + str(s) + os.linesep) out.flush() def printError(s): out.write(timeString() + ' [Error]' + ' ' + str(s) + os.linesep) out.flush() def lcg_file_download(vo,guid,localFilePath,timeout=60,maxRetry=3): cmd = 'lcg-cp -t %d --vo %s %s file://%s' % (timeout,vo,guid,localFilePath) printInfo('LFC_HOST set to %s' % os.environ['LFC_HOST']) i = 0 rc = 0 isDone = False try_again = True while try_again: i = i + 1 try: ps = os.popen(cmd) ps.close() isDone = True except IOError: printError("Download file %s from iocache failed ... trial %d." % (os.path.basename(localFilePath),i)) isDone = False if isDone: try_again = False elif i == maxRetry: try_again = False else: try_again = True return isDone ## system command executor with subprocess def execSyscmdSubprocess(cmd, wdir=os.getcwd()): import os, subprocess global exitcode outfile = file('stdout','w') errorfile = file('stderr','w') try: child = subprocess.Popen(cmd, cwd=wdir, shell=True, stdout=outfile, stderr=errorfile) while 1: exitcode = child.poll() if exitcode is not None: break else: outfile.flush() errorfile.flush() monitor.progress() time.sleep(0.3) finally: monitor.progress() outfile.flush() errorfile.flush() outfile.close() errorfile.close() return True ## system command executor with multi-thread ## stderr/stdout handler def execSyscmdEnhanced(cmd, wdir=os.getcwd()): import os, threading cwd = os.getcwd() isDone = False try: ## change to the working directory os.chdir(wdir) child = popen2.Popen3(cmd,1) child.tochild.close() # don't need stdin class PipeThread(threading.Thread): def __init__(self,infile,outfile,stopcb): self.outfile = outfile self.infile = infile self.stopcb = stopcb self.finished = 0 threading.Thread.__init__(self) def run(self): stop = False while not stop: buf = self.infile.read(10000) self.outfile.write(buf) self.outfile.flush() time.sleep(0.01) stop = self.stopcb() #FIXME: should we do here?: self.infile.read() #FIXME: this is to make sure that all the output is read (if more than buffer size of output was produced) self.finished = 1 def stopcb(poll=False): global exitcode if poll: exitcode = child.poll() return exitcode != -1 out_thread = PipeThread(child.fromchild, sys.stdout, stopcb) err_thread = PipeThread(child.childerr, sys.stderr, stopcb) out_thread.start() err_thread.start() while not out_thread.finished and not err_thread.finished: stopcb(True) monitor.progress() time.sleep(0.3) monitor.progress() sys.stdout.flush() sys.stderr.flush() isDone = True except(Exception,e): isDone = False ## return to the original directory os.chdir(cwd) return isDone ############################################################################################ """ Sandbox functions used in the job wrapper script on the worker node. The text of this module is sourced into the job wrapper script. It therefore may use ###TAGS### which are expanded in the wrapper script. """ INPUT_TARBALL_NAME = '_input_sandbox.tgz' OUTPUT_TARBALL_NAME = '_output_sandbox.tgz' PYTHON_DIR = '_python' def getPackedInputSandbox(tarpath,dest_dir='.'): """Get all sandbox_files from tarball and write them to the workdir. This function is called by wrapper script at the run time. Arguments: 'tarpath': a path to the tarball 'dest_dir': a destination directory """ #tgzfile = os.path.join(src_dir,INPUT_TARBALL_NAME) tgzfile = tarpath # ## Curent release with os module # if os.system("tar -C %s -xzf %s"%(dest_dir,tgzfile)) != 0: raise Error('cannot upack tarball with InputSandbox') # ## Future release with tarfile module # # tf = tarfile.open(tgzfile,"r:gz") # # [tf.extract(tarinfo,dest_dir) for tarinfo in tf] # # tf.close() def getInputSandbox(src_dir,dest_dir='.'): """Get all input sandbox_files from tarball and write them to the workdir. This function is called by wrapper script at the run time. Arguments: 'src_dir': a source directory with InputSandbox files. 'dest_dir': a destination directory """ os.system("tar chf - -C %s . | tar xf - -C %s" %(src_dir,dest_dir)) def createOutputSandbox(output_patterns,filter,dest_dir): """Get all files matching output patterns except filtered with filter and write them to the destination directory. This function is called by wrapper script at the run time. Arguments: 'output_patterns': list of filenames or patterns. 'filter': function to filter files (return True to except) 'dest_dir': destination directory for output files """ from Ganga.Utility.files import multi_glob,recursive_copy for f in multi_glob(output_patterns,filter): try: recursive_copy(f,dest_dir) except Exception,x: print "ERROR: (job ###JOBID### createOutput )",x def createPackedOutputSandbox(output_patterns,filter,dest_dir): """Get all files matching output patterns except filtered with filter and put them to the Sandbox tarball in destination directory. This function is called by wrapper script at the run time. Arguments: 'output_patterns': list of filenames or patterns. 'filter': function to filter files (return True to except) 'dest_dir': destination directory for tarball """ tgzfile = os.path.join(dest_dir,OUTPUT_TARBALL_NAME) from Ganga.Utility.files import multi_glob,recursive_copy outputlist = multi_glob(output_patterns,filter) # ## Curent release with os module # if len(outputlist) > 0: if os.system("tar czf %s %s"%(tgzfile," ".join(outputlist))) != 0: print "ERROR: (job ###JOBID### createPackedOutput ) can't creat tarball" # ## Future release with tarball module # # tf = tarfile.open(tgzfile,"w:gz") # tf.dereference=True # [tf.add(f) for f in outputlist] # tf.close() ############################################################################################ ## Main program ## outputsandbox = ['output_guids', 'output_location', 'output_data'] input_sandbox = {'local': ['_input_sandbox_49.tgz', '_input_sandbox_49_master.tgz'], 'remote': {}} wrapperlog = '__jobscript__.log' appexec = './athena-lcg.sh' appargs = '' exitcode=-1 import sys, stat wdir = os.getcwd() sys.path.insert(0,os.path.join(wdir,PYTHON_DIR)) os.environ['PATH'] = '.:'+os.environ['PATH'] vo = os.environ['GANGA_LCG_VO'] out = open('%s' % wrapperlog,'w') try: printInfo('Job Wrapper start.') # download inputsandbox from remote cache for f,guid in input_sandbox['remote'].iteritems(): if not lcg_file_download(vo,guid,os.path.join(wdir,f)): raise Exception('Download remote input %s:%s failed.' % (guid,f) ) else: getPackedInputSandbox(f) printInfo('Download inputsandbox from iocache passed.') # unpack inputsandbox from wdir for f in input_sandbox['local']: getPackedInputSandbox(f) printInfo('Unpack inputsandbox passed.') printInfo('Loading Python modules ...') # check the python library path try: printInfo(' ** PYTHON_DIR: %s' % os.environ['PYTHON_DIR']) except KeyError: pass try: printInfo(' ** PYTHONPATH: %s' % os.environ['PYTHONPATH']) except KeyError: pass for lib_path in sys.path: printInfo(' ** sys.path: %s' % lib_path) def createMonitoringObject(): from Ganga.Lib.MonitoringServices.Composite import CompositeMonitoringService; from Ganga.Lib.MonitoringServices.ARDADashboard.LCG.ARDADashboardLCGAthena import ARDADashboardLCGAthena; return CompositeMonitoringService([ ARDADashboardLCGAthena,],[ {'gridBackend': 'LCG', 'applicationVersion': '13.0.40', 'gangaTaskId': 'ganga_49_lester@pcfk.hep.phy.cam.ac.uk:/usera/lester/gangadir/repository/', 'VO': 'atlas', 'application': 'Athena', 'gridCertificate': '/C=UK/O=eScience/OU=Cambridge/L=UCS/CN=christopher lester/CN=proxy\n', 'activity': 'analysis', 'dataset': 'fdr08_run1.0003077.StreamMuon.merge.AOD.o1_r12_t1', 'gangaJobId': ''},]) monitor = createMonitoringObject() monitor.start() # execute application try: #try to make shipped executable executable os.chmod('%s/%s'% (wdir,appexec),stat.S_IXUSR|stat.S_IRUSR|stat.S_IWUSR) except: pass status = False try: # use subprocess to run the user's application if the module is available on the worker node import subprocess printInfo('Load application executable with subprocess module') status = execSyscmdSubprocess('%s %s' % (appexec,appargs)) except ImportError,err: # otherwise, use separate threads to control process IO pipes printInfo('Load application executable with separate threads') status = execSyscmdEnhanced('%s %s' % (appexec,appargs)) if not status: raise Exception('Application execution failed.') printInfo('Application execution passed with exit code %d.' % exitcode) createPackedOutputSandbox(outputsandbox,None,wdir) # pack outputsandbox # printInfo('== check output ==') # for line in os.popen('pwd; ls -l').readlines(): # printInfo(line) printInfo('Pack outputsandbox passed.') monitor.stop(exitcode) except Exception,e: printError(sys.exc_info()[0]) printError(sys.exc_info()[1]) str_traceback = traceback.format_tb(sys.exc_info()[2]) for str_tb in str_traceback: printError(str_tb) printInfo('Job Wrapper stop.') out.close() # always return exit code 0 so the in the case of application failure # one can always get stdout and stderr back to the UI for debug. sys.exit(0)