<s> import errno <EOL> import logging <EOL> import os <EOL> import time <EOL> from django . utils . functional import wraps <EOL> from django . utils . translation import ugettext as _ <EOL> from desktop . lib . exceptions_renderable import PopupException <EOL> from desktop . lib . i18n import smart_str <EOL> from desktop . lib . parameterization import find_variables <EOL> from desktop . models import Document2 <EOL> from hadoop import cluster <EOL> from hadoop . fs . hadoopfs import Hdfs <EOL> from oozie . utils import convert_to_server_timezone <EOL> from liboozie . oozie_api import get_oozie <EOL> from liboozie . conf import REMOTE_DEPLOYMENT_DIR <EOL> from liboozie . credentials import Credentials <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def submit_dryrun ( run_func ) : <EOL> def decorate ( self , deployment_dir = None ) : <EOL> if self . oozie_id is not None : <EOL> raise Exception ( _ ( "<STR_LIT>" ) % ( self . oozie_id , ) ) <EOL> jt_address = cluster . get_cluster_addr_for_job_submission ( ) <EOL> if deployment_dir is None : <EOL> self . _update_properties ( jt_address ) <EOL> deployment_dir = self . deploy ( ) <EOL> self . _update_properties ( jt_address , deployment_dir ) <EOL> if self . properties . get ( '<STR_LIT>' ) : <EOL> self . api . dryrun ( self . properties ) <EOL> return run_func ( self , deployment_dir ) <EOL> return wraps ( run_func ) ( decorate ) <EOL> class Submission ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , user , job = None , fs = None , jt = None , properties = None , oozie_id = None , local_tz = None ) : <EOL> self . job = job <EOL> self . user = user <EOL> self . fs = fs <EOL> self . jt = jt <EOL> self . oozie_id = oozie_id <EOL> self . api = get_oozie ( self . user ) <EOL> if properties is not None : <EOL> self . properties = properties <EOL> else : <EOL> self . properties = { } <EOL> if local_tz and isinstance ( self . job . data , dict ) : <EOL> local_tz = self . job . data . get ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in self . properties : <EOL> properties [ '<STR_LIT>' ] = convert_to_server_timezone ( self . properties [ '<STR_LIT>' ] , local_tz ) <EOL> if '<STR_LIT>' in self . properties : <EOL> properties [ '<STR_LIT>' ] = convert_to_server_timezone ( self . properties [ '<STR_LIT>' ] , local_tz ) <EOL> self . properties [ '<STR_LIT>' ] = self . api . security_enabled <EOL> def __str__ ( self ) : <EOL> if self . oozie_id : <EOL> res = "<STR_LIT>" % ( self . oozie_id , ) <EOL> else : <EOL> res = "<STR_LIT>" % ( self . job . name , self . job . id , self . user ) <EOL> if self . oozie_id : <EOL> res += "<STR_LIT>" + self . oozie_id <EOL> return res <EOL> @ submit_dryrun <EOL> def run ( self , deployment_dir = None ) : <EOL> """<STR_LIT>""" <EOL> self . oozie_id = self . api . submit_job ( self . properties ) <EOL> LOG . info ( "<STR_LIT>" % ( self , ) ) <EOL> if self . _is_workflow ( ) : <EOL> self . api . job_control ( self . oozie_id , '<STR_LIT:start>' ) <EOL> LOG . info ( "<STR_LIT>" % ( self , ) ) <EOL> return self . oozie_id <EOL> def rerun ( self , deployment_dir , fail_nodes = None , skip_nodes = None ) : <EOL> jt_address = cluster . get_cluster_addr_for_job_submission ( ) <EOL> self . _update_properties ( jt_address , deployment_dir ) <EOL> self . properties . update ( { '<STR_LIT>' : deployment_dir } ) <EOL> if fail_nodes : <EOL> self . properties . update ( { '<STR_LIT>' : fail_nodes } ) <EOL> elif not skip_nodes : <EOL> self . properties . update ( { '<STR_LIT>' : '<STR_LIT:false>' } ) <EOL> else : <EOL> self . properties . update ( { '<STR_LIT>' : skip_nodes } ) <EOL> self . api . rerun ( self . oozie_id , properties = self . properties ) <EOL> LOG . info ( "<STR_LIT>" % ( self , ) ) <EOL> return self . oozie_id <EOL> def rerun_coord ( self , deployment_dir , params ) : <EOL> jt_address = cluster . get_cluster_addr_for_job_submission ( ) <EOL> self . _update_properties ( jt_address , deployment_dir ) <EOL> self . properties . update ( { '<STR_LIT>' : deployment_dir } ) <EOL> self . api . job_control ( self . oozie_id , action = '<STR_LIT>' , properties = self . properties , parameters = params ) <EOL> LOG . info ( "<STR_LIT>" % ( self , ) ) <EOL> return self . oozie_id <EOL> def update_coord ( self ) : <EOL> self . api = get_oozie ( self . user , api_version = "<STR_LIT>" ) <EOL> self . api . job_control ( self . oozie_id , action = '<STR_LIT>' , properties = self . properties , parameters = None ) <EOL> LOG . info ( "<STR_LIT>" % ( self , ) ) <EOL> return self . oozie_id <EOL> def rerun_bundle ( self , deployment_dir , params ) : <EOL> jt_address = cluster . get_cluster_addr_for_job_submission ( ) <EOL> self . _update_properties ( jt_address , deployment_dir ) <EOL> self . properties . update ( { '<STR_LIT>' : deployment_dir } ) <EOL> self . api . job_control ( self . oozie_id , action = '<STR_LIT>' , properties = self . properties , parameters = params ) <EOL> LOG . info ( "<STR_LIT>" % ( self , ) ) <EOL> return self . oozie_id <EOL> def deploy ( self ) : <EOL> try : <EOL> deployment_dir = self . _create_deployment_dir ( ) <EOL> except Exception , ex : <EOL> msg = _ ( "<STR_LIT>" % ex ) <EOL> LOG . exception ( msg ) <EOL> raise PopupException ( message = msg , detail = str ( ex ) ) <EOL> if self . api . security_enabled : <EOL> jt_address = cluster . get_cluster_addr_for_job_submission ( ) <EOL> self . _update_properties ( jt_address ) <EOL> if hasattr ( self . job , '<STR_LIT>' ) : <EOL> for action in self . job . nodes : <EOL> if action . data [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> from oozie . models2 import Workflow <EOL> workflow = Workflow ( document = Document2 . objects . get_by_uuid ( user = self . user , uuid = action . data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) ) <EOL> sub_deploy = Submission ( self . user , workflow , self . fs , self . jt , self . properties ) <EOL> workspace = sub_deploy . deploy ( ) <EOL> self . job . override_subworkflow_id ( action , workflow . id ) <EOL> self . properties [ '<STR_LIT>' % workflow . uuid ] = workspace <EOL> elif action . data [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> from notebook . models import Notebook <EOL> notebook = Notebook ( document = Document2 . objects . get_by_uuid ( user = self . user , uuid = action . data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) ) <EOL> self . _create_file ( deployment_dir , action . data [ '<STR_LIT:name>' ] + '<STR_LIT>' , notebook . get_str ( ) ) <EOL> oozie_xml = self . job . to_xml ( self . properties ) <EOL> self . _do_as ( self . user . username , self . _copy_files , deployment_dir , oozie_xml , self . properties ) <EOL> return deployment_dir <EOL> def get_external_parameters ( self , application_path ) : <EOL> """<STR_LIT>""" <EOL> deployment_dir = os . path . dirname ( application_path ) <EOL> xml = self . fs . do_as_user ( self . user , self . fs . read , application_path , <NUM_LIT:0> , <NUM_LIT:1> * <NUM_LIT> ** <NUM_LIT:2> ) <EOL> properties_file = deployment_dir + '<STR_LIT>' <EOL> if self . fs . do_as_user ( self . user , self . fs . exists , properties_file ) : <EOL> properties = self . fs . do_as_user ( self . user , self . fs . read , properties_file , <NUM_LIT:0> , <NUM_LIT:1> * <NUM_LIT> ** <NUM_LIT:2> ) <EOL> else : <EOL> properties = None <EOL> return self . _get_external_parameters ( xml , properties ) <EOL> def _get_external_parameters ( self , xml , properties = None ) : <EOL> from oozie . models import DATASET_FREQUENCY <EOL> parameters = dict ( [ ( var , '<STR_LIT>' ) for var in find_variables ( xml , include_named = False ) if not self . _is_coordinator ( ) or var not in DATASET_FREQUENCY ] ) <EOL> if properties : <EOL> parameters . update ( dict ( [ line . strip ( ) . split ( '<STR_LIT:=>' ) <EOL> for line in properties . split ( '<STR_LIT:\n>' ) if not line . startswith ( '<STR_LIT:#>' ) and len ( line . strip ( ) . split ( '<STR_LIT:=>' ) ) == <NUM_LIT:2> ] ) ) <EOL> return parameters <EOL> def _update_properties ( self , jobtracker_addr , deployment_dir = None ) : <EOL> LOG . info ( '<STR_LIT>' % ( self . fs , self . jt ) ) <EOL> if self . jt and self . jt . logical_name : <EOL> jobtracker_addr = self . jt . logical_name <EOL> if self . fs . logical_name : <EOL> fs_defaultfs = self . fs . logical_name <EOL> else : <EOL> fs_defaultfs = self . fs . fs_defaultfs <EOL> self . properties . update ( { <EOL> '<STR_LIT>' : jobtracker_addr , <EOL> '<STR_LIT>' : fs_defaultfs , <EOL> } ) <EOL> if self . job and deployment_dir : <EOL> self . properties . update ( { <EOL> self . job . PROPERTY_APP_PATH : self . fs . get_hdfs_path ( deployment_dir ) , <EOL> self . job . HUE_ID : self . job . id <EOL> } ) <EOL> if self . api . security_enabled : <EOL> credentials = Credentials ( ) <EOL> credentials . fetch ( self . api ) <EOL> self . properties [ '<STR_LIT>' ] = credentials . get_properties ( ) <EOL> def _create_deployment_dir ( self ) : <EOL> """<STR_LIT>""" <EOL> create_directories ( self . fs ) <EOL> if self . user != self . job . document . owner : <EOL> path = REMOTE_DEPLOYMENT_DIR . get ( ) . replace ( '<STR_LIT>' , self . user . username ) . replace ( '<STR_LIT>' , str ( time . time ( ) ) ) . replace ( '<STR_LIT>' , str ( self . job . id ) ) <EOL> if self . fs . exists ( self . job . deployment_dir ) : <EOL> self . fs . copy_remote_dir ( self . job . deployment_dir , path , owner = self . user ) <EOL> else : <EOL> self . _create_dir ( path ) <EOL> else : <EOL> path = self . job . deployment_dir <EOL> self . _create_dir ( path ) <EOL> return path <EOL> def _create_dir ( self , path , perms = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> statbuf = self . fs . stats ( path ) <EOL> if not statbuf . isDir : <EOL> msg = _ ( "<STR_LIT>" ) % ( path , ) <EOL> LOG . error ( msg ) <EOL> raise Exception ( msg ) <EOL> except IOError , ex : <EOL> if ex . errno != errno . ENOENT : <EOL> msg = _ ( "<STR_LIT>" ) % ( path , ex ) <EOL> LOG . exception ( msg ) <EOL> raise IOError ( ex . errno , msg ) <EOL> if not self . fs . exists ( path ) : <EOL> self . _do_as ( self . user . username , self . fs . mkdir , path , perms ) <EOL> if perms is not None : <EOL> self . _do_as ( self . user . username , self . fs . chmod , path , perms ) <EOL> return path <EOL> def _copy_files ( self , deployment_dir , oozie_xml , oozie_properties ) : <EOL> """<STR_LIT>""" <EOL> self . _create_file ( deployment_dir , self . job . XML_FILE_NAME , oozie_xml ) <EOL> self . _create_file ( deployment_dir , '<STR_LIT>' , data = '<STR_LIT:\n>' . join ( [ '<STR_LIT>' % ( key , val ) for key , val in oozie_properties . iteritems ( ) ] ) ) <EOL> files = [ ] <EOL> lib_path = self . fs . join ( deployment_dir , '<STR_LIT>' ) <EOL> if hasattr ( self . job , '<STR_LIT>' ) : <EOL> for node in self . job . nodes : <EOL> jar_path = node . data [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> if jar_path : <EOL> if not jar_path . startswith ( '<STR_LIT:/>' ) : <EOL> jar_path = self . fs . join ( self . job . deployment_dir , jar_path ) <EOL> if not jar_path . startswith ( lib_path ) : <EOL> files . append ( jar_path ) <EOL> if files : <EOL> for jar_file in files : <EOL> LOG . debug ( "<STR_LIT>" % jar_file ) <EOL> jar_lib_path = self . fs . join ( lib_path , self . fs . basename ( jar_file ) ) <EOL> if self . fs . exists ( jar_lib_path ) and self . fs . exists ( jar_file ) : <EOL> stat_src = self . fs . stats ( jar_file ) <EOL> stat_dest = self . fs . stats ( jar_lib_path ) <EOL> if stat_src . fileId != stat_dest . fileId : <EOL> self . fs . remove ( jar_lib_path , skip_trash = True ) <EOL> self . fs . copyfile ( jar_file , jar_lib_path ) <EOL> def _do_as ( self , username , fn , * args , ** kwargs ) : <EOL> prev_user = self . fs . user <EOL> try : <EOL> self . fs . setuser ( username ) <EOL> return fn ( * args , ** kwargs ) <EOL> finally : <EOL> self . fs . setuser ( prev_user ) <EOL> def remove_deployment_dir ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> path = self . job . deployment_dir <EOL> if self . _do_as ( self . user . username , self . fs . exists , path ) : <EOL> self . _do_as ( self . user . username , self . fs . rmtree , path ) <EOL> except Exception , ex : <EOL> LOG . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> self . job . name , self . user , ex ) <EOL> def _is_workflow ( self ) : <EOL> from oozie . models2 import Workflow <EOL> return Workflow . PROPERTY_APP_PATH in self . properties <EOL> def _is_coordinator ( self ) : <EOL> from oozie . models2 import Coordinator <EOL> return Coordinator . PROPERTY_APP_PATH in self . properties <EOL> def _create_file ( self , deployment_dir , file_name , data , do_as = False ) : <EOL> file_path = self . fs . join ( deployment_dir , file_name ) <EOL> if do_as : <EOL> self . fs . do_as_user ( self . user , self . fs . create , file_path , overwrite = True , permission = <NUM_LIT:0> <NUM_LIT> , data = smart_str ( data ) ) <EOL> else : <EOL> self . fs . create ( file_path , overwrite = True , permission = <NUM_LIT:0> <NUM_LIT> , data = smart_str ( data ) ) <EOL> LOG . debug ( "<STR_LIT>" % ( file_path , ) ) <EOL> def create_directories ( fs , directory_list = [ ] ) : <EOL> directories = [ REMOTE_DEPLOYMENT_DIR . get ( ) ] + directory_list <EOL> for directory in directories : <EOL> if not fs . do_as_user ( fs . DEFAULT_USER , fs . exists , directory ) : <EOL> remote_home_dir = Hdfs . join ( '<STR_LIT>' , fs . DEFAULT_USER ) <EOL> if directory . startswith ( remote_home_dir ) : <EOL> fs . do_as_user ( fs . DEFAULT_USER , fs . create_home_dir , remote_home_dir ) <EOL> fs . do_as_user ( fs . DEFAULT_USER , fs . mkdir , directory , <NUM_LIT:0> <NUM_LIT> ) <EOL> fs . do_as_user ( fs . DEFAULT_USER , fs . chmod , directory , <NUM_LIT:0> <NUM_LIT> ) </s>
