#! /usr/bin/env python ############################################################# # # # Author: Bertrand Neron # # Organization:'Biological Software and Databases' Group, # # Institut Pasteur, Paris. # # Distributed under GPLv2 Licence. Please refer to the # # COPYING.LIB document. # # # ############################################################# import os , sys MOBYLEHOME = None if os.environ.has_key('MOBYLEHOME'): MOBYLEHOME = os.environ['MOBYLEHOME'] if not MOBYLEHOME: sys.exit( 'MOBYLEHOME must be defined in your environment' ) if ( os.path.join( MOBYLEHOME , 'Src' ) ) not in sys.path: sys.path.append( os.path.join( MOBYLEHOME , 'Src' ) ) import shutil import re from lxml import etree from Mobyle.ConfigManager import Config from Mobyle.JobState import JobState from Mobyle.Registry import registry from Mobyle.MobyleError import MobyleError class JobUpdater(object): def __init__(self , config , logger ): self.cfg = config self.log = logger self.skipped_jobs = [] self.skipped_services = [] self.jobs_url = self.cfg.results_url() def _on_error(self , err ): self.log.warning( "%s : %s" ( err.filename , err )) def update( self , repository , force = False , dry_run = False , status_only = False ): skipped_jobs = [] jobs_paths = os.walk( repository , onerror = self._on_error ) current_service = '' for path in jobs_paths: job_path = path[0] if 'index.xml' in path[2]: if not '.admin' in path[2]: self.log.warning( "the directory %s have an index.xlm but not .admin may be not a job : skip" %path[2] ) skipped_jobs.append( job_path ) continue service_name , job_key = os.path.split( path[0] ) service_name = os.path.basename( service_name ) if service_name != current_service: current_service = service_name self.log.info( "-------- enter service %s --------" % service_name) self.log.info( "====================" ) self.log.info( "updating job %s" %job_key ) job_dir_mtime = os.path.getmtime( job_path ) job_dir_atime = os.path.getatime( job_path ) index_mtime = os.path.getmtime( os.path.join( job_path , 'index.xml') ) index_atime = os.path.getatime( os.path.join( job_path , 'index.xml') ) index_path = os.path.join( job_path , 'index.xml') try: os.utime( job_path , ( job_dir_atime , job_dir_mtime ) ) except Exception , err : self.log.warning( "cannot modify the mtime of %s : %s : fix this problem before rerun updater" %(job_path , err ) ) continue bckp_path = self.back_up(job_path, index_path, dry_run) try: if not status_only: try: self.fix_definition( job_path , service_name , job_key , index_path , bckp_path , force = force , dry_run = dry_run ) except Exception, err: self.roll_back(bckp_path, index_path) try: self.fix_status( job_path , index_path , force = force , dry_run = dry_run ) except Exception, err: self.log.warning( "Error occured during status fixing : %s" % err, exc_info = True ) self.roll_back(bckp_path, index_path) finally: self.set_time(index_path, index_atime, index_mtime, job_path, job_dir_atime, job_dir_mtime) def fix_definition( self , job_path , service_name , job_key ,index_path , bckp_path ,force = False ,dry_run = False): self.log.info( "fixing %s job definition" % job_path ) try: js = JobState( uri = job_path ) except Exception , err: self.log.warning( "cannot load job %s : %s : skipped" %( job_path , err ) ) return if js.getDefinition() is not None and not force: self.log.warning("the job %s has already a definition: skipped " % job_path ) return new_program_def_url = registry.getProgramUrl( service_name , 'local' ) self.log.info( "registry.getProgramUrl( %s , 'local' ) = %s "% (service_name , new_program_def_url ) ) new_host = self.cfg.root_url() self.log.info( "new host = %s "% new_host ) job_id = js.getID() new_job_id = "%s/%s" %(self.jobs_url , re.search( '.*/(.*/.*)$' , job_id ).group(1) ) self.log.info( "new job id = %s "% new_job_id ) if not dry_run: js.setName( new_program_def_url ) try: js.setDefinition( new_program_def_url) except MobyleError: self.log.warning( "the service : local.%s is not deployed to update job %s, deploy %s before" %(service_name , job_path, service_name) ) return js.setHost( new_host ) js.setID( new_job_id ) try: js.commit() except Exception ,err : self.log.warning( "cannot commit updated job %s : %s " %( job_path , err ) ) self.roll_back(bckp_path, index_path) else: pass def fix_status(self , job_path , index_path , force = False , dry_run = False ): self.log.info( "fixing %s job status" % job_path ) status_path = os.path.join( job_path , 'mobyle_status.xml') if os.path.exists( status_path ) and not force: self.log.info("%s already exist : skip" % status_path ) return else: doc = etree.parse( os.path.join( job_path , 'index.xml') ) root = doc.getroot() status_node = root.find( 'status' ) status = status_node.find( 'value' ).text if status in ( "submitted", "pending", "running", "hold"): return try: status_file = open( status_path , 'w' ) status_file.write( etree.tostring( status_node , pretty_print = True , encoding='UTF-8' )) except Exception , err : self.log.error( 'cannot make mobyle_status.xml : %s' % err ) finally: try: status_file.close() except: pass root.remove( status_node ) try: tmp_file = open( os.path.join( job_path ,'tmp_index.xml' ) , 'w' ) tmp_file.write( etree.tostring( doc , pretty_print = True , encoding='UTF-8' )) tmp_file.close() os.rename( tmp_file.name , index_path ) except Exception , err : self.log.error( 'cannot make index.xml : %s' % err ) def back_up(self , job_path , index_path , dry_run = False ): bckp_path = os.path.join( job_path , '.index.xml.bckp') if not os.path.exists( bckp_path ): self.log.debug( "make %s back up" % job_path ) if not dry_run: try: shutil.copy( index_path , bckp_path ) except Exception , err: self.log.warning( "cannot make %s back up " % index_path) return bckp_path def roll_back(self , bckp_path , index_path ): try: shutil.copy( bckp_path , index_path ) except Exception: self.log.warning( "cannot roll back ") return try: os.unlink( bckp_path ) except Exception: self.log.warning( "cannot remove back up index % " % bckp_path ) return self.log.warning( "job %s has been rolled back " %( os.path.dirname( index_path ) ) ) return def set_time(self , index_path , index_atime , index_mtime , job_path , job_dir_atime , job_dir_mtime ): os.utime( index_path , ( index_atime , index_mtime ) ) os.utime( job_path , ( job_dir_atime , job_dir_mtime ) ) if __name__ == "__main__": from optparse import OptionParser parser = OptionParser( ) parser.add_option( "-s" , "--services", action="store", type = 'string', dest="services", help="comma separated list of SERVICES to update." ) parser.add_option( "-l" , "--log", action="store", type = 'string', dest = "log_file", help= "Path to the Logfile where put the logs.") parser.add_option( "-n" , "--dry-run", action="store_true", dest = "dry_run", default = False , help= "don't actually do anything.") parser.add_option( "-f" , "--force", action="store_true", dest = "force", default = False , help= "force to update job even if it was already updated.") parser.add_option("-v", "--verbose", action="count", dest="verbosity", default= 0, help="increase the verbosity level. There is 4 levels: Error messages (default), Warning (-v), Info (-vv) and Debug.(-vvv)") parser.add_option("--status-only", action="store_true", dest="status_only", default= False, help="fix the job status only") options, args = parser.parse_args() import logging logger = logging.getLogger( 'jobUpdater' ) if options.log_file is None: handler = logging.StreamHandler( sys.stderr ) else: try: handler = logging.FileHandler( options.log_file , 'w' ) except(IOError , OSError) , err: print >> sys.stderr , "cannot log messages in %s: %s"%( options.log_file , err ) sys.exit(1) if options.verbosity < 2: formatter = logging.Formatter( '%(filename)-10s : %(levelname)-8s : %(asctime)s : %(message)s' , '%a, %d %b %Y %H:%M:%S' ) else: formatter = logging.Formatter( '%(filename)-10s : %(levelname)-8s : L %(lineno)d : %(asctime)s : %(message)s' , '%a, %d %b %Y %H:%M:%S' ) handler.setFormatter( formatter ) logger.addHandler( handler) if options.verbosity == 0: logger.setLevel( logging.WARNING ) elif options.verbosity == 1: logger.setLevel( logging.INFO ) else: logger.setLevel( logging.DEBUG ) config = Config() job_updater = JobUpdater( config , logger ) jobs_repository = config.results_path() if options.services is None: job_updater.update( jobs_repository , force = options.force , dry_run = options.dry_run , status_only = options.status_only ) else: services = options.services.split( ',' ) for service in services: service_path = os.path.join( jobs_repository , service ) job_updater.update( service_path , force = options.force , dry_run = options.dry_run , status_only = options.status_only )