<s> import abc <EOL> import os <EOL> import uuid <EOL> import xml . dom . minidom as xml <EOL> from oslo_config import cfg <EOL> import six <EOL> from sahara import conductor as c <EOL> from sahara import context <EOL> from sahara . service . edp import base_engine <EOL> from sahara . service . edp . binary_retrievers import dispatch <EOL> from sahara . service . edp import hdfs_helper as h <EOL> from sahara . service . edp import job_utils <EOL> from sahara . service . edp . oozie import oozie as o <EOL> from sahara . service . edp . oozie . workflow_creator import workflow_factory <EOL> from sahara . service . validations . edp import job_execution as j <EOL> from sahara . utils import edp <EOL> from sahara . utils import remote <EOL> from sahara . utils import xmlutils as x <EOL> CONF = cfg . CONF <EOL> conductor = c . API <EOL> @ six . add_metaclass ( abc . ABCMeta ) <EOL> class OozieJobEngine ( base_engine . JobEngine ) : <EOL> def __init__ ( self , cluster ) : <EOL> self . cluster = cluster <EOL> self . plugin = job_utils . get_plugin ( self . cluster ) <EOL> def _get_client ( self ) : <EOL> return o . OozieClient ( self . get_oozie_server_uri ( self . cluster ) , <EOL> self . get_oozie_server ( self . cluster ) ) <EOL> def _get_oozie_job_params ( self , hdfs_user , path_to_workflow , <EOL> oozie_params , use_hbase_lib , <EOL> scheduled_params = None , job_dir = None , <EOL> job_execution_type = None ) : <EOL> oozie_libpath_key = "<STR_LIT>" <EOL> oozie_libpath = "<STR_LIT>" <EOL> rm_path = self . get_resource_manager_uri ( self . cluster ) <EOL> nn_path = self . get_name_node_uri ( self . cluster ) <EOL> hbase_common_lib_path = "<STR_LIT>" % ( nn_path , h . HBASE_COMMON_LIB_PATH ) <EOL> if use_hbase_lib : <EOL> if oozie_libpath_key in oozie_params : <EOL> oozie_libpath = "<STR_LIT>" % ( oozie_params . get ( oozie_libpath_key , <EOL> "<STR_LIT>" ) , hbase_common_lib_path ) <EOL> else : <EOL> oozie_libpath = hbase_common_lib_path <EOL> if job_execution_type == "<STR_LIT>" : <EOL> app_path = "<STR_LIT>" <EOL> job_parameters = { <EOL> "<STR_LIT:start>" : scheduled_params . get ( '<STR_LIT:start>' ) , <EOL> "<STR_LIT:end>" : scheduled_params . get ( '<STR_LIT:end>' ) , <EOL> "<STR_LIT>" : scheduled_params . get ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" : "<STR_LIT>" % ( nn_path , job_dir ) , <EOL> app_path : "<STR_LIT>" % ( nn_path , job_dir ) } <EOL> else : <EOL> app_path = "<STR_LIT>" <EOL> job_parameters = { <EOL> app_path : "<STR_LIT>" % ( nn_path , path_to_workflow ) } <EOL> job_parameters [ "<STR_LIT>" ] = nn_path <EOL> job_parameters [ "<STR_LIT>" ] = hdfs_user <EOL> job_parameters [ "<STR_LIT>" ] = rm_path <EOL> job_parameters [ oozie_libpath_key ] = oozie_libpath <EOL> job_parameters [ "<STR_LIT>" ] = "<STR_LIT:true>" <EOL> if app_path in oozie_params : <EOL> del oozie_params [ app_path ] <EOL> if oozie_libpath_key in oozie_params : <EOL> del oozie_params [ oozie_libpath_key ] <EOL> job_parameters . update ( oozie_params ) <EOL> return job_parameters <EOL> def _upload_workflow_file ( self , where , job_dir , wf_xml , hdfs_user ) : <EOL> with remote . get_remote ( where ) as r : <EOL> h . put_file_to_hdfs ( r , wf_xml , "<STR_LIT>" , job_dir , hdfs_user ) <EOL> return "<STR_LIT>" % job_dir <EOL> def _upload_coordinator_file ( self , where , job_dir , wf_xml , hdfs_user ) : <EOL> with remote . get_remote ( where ) as r : <EOL> h . put_file_to_hdfs ( r , wf_xml , "<STR_LIT>" , job_dir , <EOL> hdfs_user ) <EOL> return "<STR_LIT>" % job_dir <EOL> def cancel_job ( self , job_execution ) : <EOL> if job_execution . engine_job_id is not None : <EOL> client = self . _get_client ( ) <EOL> client . kill_job ( job_execution ) <EOL> return client . get_job_info ( job_execution ) <EOL> def get_job_status ( self , job_execution ) : <EOL> if job_execution . engine_job_id is not None : <EOL> return self . _get_client ( ) . get_job_info ( job_execution ) <EOL> def _prepare_run_job ( self , job_execution ) : <EOL> ctx = context . ctx ( ) <EOL> data_source_urls = { } <EOL> prepared_job_params = { } <EOL> job = conductor . job_get ( ctx , job_execution . job_id ) <EOL> input_source , output_source = job_utils . get_data_sources ( <EOL> job_execution , job , data_source_urls , self . cluster ) <EOL> additional_sources , updated_job_configs = ( <EOL> job_utils . resolve_data_source_references ( job_execution . job_configs , <EOL> job_execution . id , <EOL> data_source_urls , <EOL> self . cluster ) <EOL> ) <EOL> job_execution = conductor . job_execution_update ( <EOL> ctx , job_execution , <EOL> { "<STR_LIT>" : job_utils . to_url_dict ( data_source_urls ) } ) <EOL> data_source_urls = job_utils . to_url_dict ( data_source_urls , <EOL> runtime = True ) <EOL> proxy_configs = updated_job_configs . get ( '<STR_LIT>' ) <EOL> configs = updated_job_configs . get ( '<STR_LIT>' , { } ) <EOL> use_hbase_lib = configs . get ( '<STR_LIT>' , { } ) <EOL> oozie_params = { } <EOL> for k in list ( configs ) : <EOL> if k . startswith ( '<STR_LIT>' ) : <EOL> oozie_params [ k ] = configs [ k ] <EOL> for data_source in [ input_source , output_source ] + additional_sources : <EOL> if data_source and data_source . type == '<STR_LIT>' : <EOL> h . configure_cluster_for_hdfs ( <EOL> self . cluster , data_source_urls [ data_source . id ] ) <EOL> break <EOL> external_hdfs_urls = self . _resolve_external_hdfs_urls ( <EOL> job_execution . job_configs ) <EOL> for url in external_hdfs_urls : <EOL> h . configure_cluster_for_hdfs ( self . cluster , url ) <EOL> hdfs_user = self . get_hdfs_user ( ) <EOL> oozie_server = self . get_oozie_server ( self . cluster ) <EOL> wf_dir = self . _create_hdfs_workflow_dir ( oozie_server , job ) <EOL> self . _upload_job_files_to_hdfs ( oozie_server , wf_dir , job , configs , <EOL> proxy_configs ) <EOL> wf_xml = workflow_factory . get_workflow_xml ( <EOL> job , self . cluster , updated_job_configs , <EOL> input_source , output_source , <EOL> hdfs_user , data_source_urls ) <EOL> path_to_workflow = self . _upload_workflow_file ( oozie_server , wf_dir , <EOL> wf_xml , hdfs_user ) <EOL> prepared_job_params [ '<STR_LIT>' ] = ctx <EOL> prepared_job_params [ '<STR_LIT>' ] = hdfs_user <EOL> prepared_job_params [ '<STR_LIT>' ] = path_to_workflow <EOL> prepared_job_params [ '<STR_LIT>' ] = use_hbase_lib <EOL> prepared_job_params [ '<STR_LIT>' ] = job_execution <EOL> prepared_job_params [ '<STR_LIT>' ] = oozie_params <EOL> prepared_job_params [ '<STR_LIT>' ] = wf_dir <EOL> prepared_job_params [ '<STR_LIT>' ] = oozie_server <EOL> return prepared_job_params <EOL> def run_job ( self , job_execution ) : <EOL> prepared_job_params = self . _prepare_run_job ( job_execution ) <EOL> path_to_workflow = prepared_job_params [ '<STR_LIT>' ] <EOL> hdfs_user = prepared_job_params [ '<STR_LIT>' ] <EOL> oozie_params = prepared_job_params [ '<STR_LIT>' ] <EOL> use_hbase_lib = prepared_job_params [ '<STR_LIT>' ] <EOL> ctx = prepared_job_params [ '<STR_LIT>' ] <EOL> job_execution = prepared_job_params [ '<STR_LIT>' ] <EOL> job_params = self . _get_oozie_job_params ( hdfs_user , <EOL> path_to_workflow , <EOL> oozie_params , <EOL> use_hbase_lib ) <EOL> client = self . _get_client ( ) <EOL> oozie_job_id = client . add_job ( x . create_hadoop_xml ( job_params ) , <EOL> job_execution ) <EOL> job_execution = conductor . job_execution_get ( ctx , job_execution . id ) <EOL> if job_execution . info [ '<STR_LIT:status>' ] == edp . JOB_STATUS_TOBEKILLED : <EOL> return ( None , edp . JOB_STATUS_KILLED , None ) <EOL> conductor . job_execution_update ( <EOL> context . ctx ( ) , job_execution . id , <EOL> { '<STR_LIT:info>' : { '<STR_LIT:status>' : edp . JOB_STATUS_READYTORUN } , <EOL> '<STR_LIT>' : oozie_job_id } ) <EOL> client . run_job ( job_execution , oozie_job_id ) <EOL> try : <EOL> status = client . get_job_info ( job_execution , oozie_job_id ) [ '<STR_LIT:status>' ] <EOL> except Exception : <EOL> status = None <EOL> return ( oozie_job_id , status , None ) <EOL> def run_scheduled_job ( self , job_execution ) : <EOL> prepared_job_params = self . _prepare_run_job ( job_execution ) <EOL> oozie_server = prepared_job_params [ '<STR_LIT>' ] <EOL> wf_dir = prepared_job_params [ '<STR_LIT>' ] <EOL> hdfs_user = prepared_job_params [ '<STR_LIT>' ] <EOL> oozie_params = prepared_job_params [ '<STR_LIT>' ] <EOL> use_hbase_lib = prepared_job_params [ '<STR_LIT>' ] <EOL> ctx = prepared_job_params [ '<STR_LIT>' ] <EOL> job_execution = prepared_job_params [ '<STR_LIT>' ] <EOL> coord_configs = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> coord_xml = self . _create_coordinator_xml ( coord_configs ) <EOL> self . _upload_coordinator_file ( oozie_server , wf_dir , coord_xml , <EOL> hdfs_user ) <EOL> job_params = self . _get_oozie_job_params ( <EOL> hdfs_user , None , oozie_params , use_hbase_lib , <EOL> job_execution . job_configs . job_execution_info , wf_dir , <EOL> "<STR_LIT>" ) <EOL> client = self . _get_client ( ) <EOL> oozie_job_id = client . add_job ( x . create_hadoop_xml ( job_params ) , <EOL> job_execution ) <EOL> job_execution = conductor . job_execution_get ( ctx , job_execution . id ) <EOL> if job_execution . info [ '<STR_LIT:status>' ] == edp . JOB_STATUS_TOBEKILLED : <EOL> return ( None , edp . JOB_STATUS_KILLED , None ) <EOL> try : <EOL> status = client . get_job_status ( job_execution , <EOL> oozie_job_id ) [ '<STR_LIT:status>' ] <EOL> except Exception : <EOL> status = None <EOL> return ( oozie_job_id , status , None ) <EOL> @ abc . abstractmethod <EOL> def get_hdfs_user ( self ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def create_hdfs_dir ( self , remote , dir_name ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def get_oozie_server_uri ( self , cluster ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def get_oozie_server ( self , cluster ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def get_name_node_uri ( self , cluster ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def get_resource_manager_uri ( self , cluster ) : <EOL> pass <EOL> def validate_job_execution ( self , cluster , job , data ) : <EOL> if job . type == edp . JOB_TYPE_SHELL : <EOL> return <EOL> if job . type == edp . JOB_TYPE_JAVA : <EOL> j . check_main_class_present ( data , job ) <EOL> else : <EOL> j . check_data_sources ( data , job ) <EOL> job_type , subtype = edp . split_job_type ( job . type ) <EOL> if job_type == edp . JOB_TYPE_MAPREDUCE and ( <EOL> subtype == edp . JOB_SUBTYPE_STREAMING ) : <EOL> j . check_streaming_present ( data , job ) <EOL> @ staticmethod <EOL> def get_possible_job_config ( job_type ) : <EOL> return workflow_factory . get_possible_job_config ( job_type ) <EOL> @ staticmethod <EOL> def get_supported_job_types ( ) : <EOL> return [ edp . JOB_TYPE_HIVE , <EOL> edp . JOB_TYPE_JAVA , <EOL> edp . JOB_TYPE_MAPREDUCE , <EOL> edp . JOB_TYPE_MAPREDUCE_STREAMING , <EOL> edp . JOB_TYPE_PIG , <EOL> edp . JOB_TYPE_SHELL ] <EOL> def _upload_job_files_to_hdfs ( self , where , job_dir , job , configs , <EOL> proxy_configs = None ) : <EOL> mains = job . mains or [ ] <EOL> libs = job . libs or [ ] <EOL> builtin_libs = edp . get_builtin_binaries ( job , configs ) <EOL> uploaded_paths = [ ] <EOL> hdfs_user = self . get_hdfs_user ( ) <EOL> job_dir_suffix = '<STR_LIT>' if job . type != edp . JOB_TYPE_SHELL else '<STR_LIT>' <EOL> lib_dir = os . path . join ( job_dir , job_dir_suffix ) <EOL> with remote . get_remote ( where ) as r : <EOL> for main in mains : <EOL> raw_data = dispatch . get_raw_binary ( <EOL> main , proxy_configs = proxy_configs , remote = r ) <EOL> if isinstance ( raw_data , dict ) and raw_data [ "<STR_LIT:type>" ] == "<STR_LIT:path>" : <EOL> h . copy_from_local ( r , raw_data [ '<STR_LIT:path>' ] , <EOL> job_dir , hdfs_user ) <EOL> else : <EOL> h . put_file_to_hdfs ( r , raw_data , main . name , <EOL> job_dir , hdfs_user ) <EOL> uploaded_paths . append ( job_dir + '<STR_LIT:/>' + main . name ) <EOL> if len ( libs ) and job_dir_suffix : <EOL> self . create_hdfs_dir ( r , lib_dir ) <EOL> for lib in libs : <EOL> raw_data = dispatch . get_raw_binary ( <EOL> lib , proxy_configs = proxy_configs , remote = remote ) <EOL> if isinstance ( raw_data , dict ) and raw_data [ "<STR_LIT:type>" ] == "<STR_LIT:path>" : <EOL> h . copy_from_local ( r , raw_data [ '<STR_LIT:path>' ] , <EOL> lib_dir , hdfs_user ) <EOL> else : <EOL> h . put_file_to_hdfs ( r , raw_data , lib . name , <EOL> lib_dir , hdfs_user ) <EOL> uploaded_paths . append ( lib_dir + '<STR_LIT:/>' + lib . name ) <EOL> for lib in builtin_libs : <EOL> h . put_file_to_hdfs ( r , lib [ '<STR_LIT>' ] , lib [ '<STR_LIT:name>' ] , lib_dir , <EOL> hdfs_user ) <EOL> uploaded_paths . append ( lib_dir + '<STR_LIT:/>' + lib [ '<STR_LIT:name>' ] ) <EOL> return uploaded_paths <EOL> def _create_hdfs_workflow_dir ( self , where , job ) : <EOL> constructed_dir = '<STR_LIT>' % self . get_hdfs_user ( ) <EOL> constructed_dir = self . _add_postfix ( constructed_dir ) <EOL> constructed_dir += '<STR_LIT>' % ( job . name , six . text_type ( uuid . uuid4 ( ) ) ) <EOL> with remote . get_remote ( where ) as r : <EOL> self . create_hdfs_dir ( r , constructed_dir ) <EOL> return constructed_dir <EOL> def _create_coordinator_xml ( self , coord_configs , config_filter = None , <EOL> appname = '<STR_LIT>' ) : <EOL> doc = xml . Document ( ) <EOL> coord = doc . createElement ( '<STR_LIT>' ) <EOL> coord . attributes [ '<STR_LIT:name>' ] = appname <EOL> coord . attributes [ '<STR_LIT:start>' ] = "<STR_LIT>" <EOL> coord . attributes [ '<STR_LIT:end>' ] = "<STR_LIT>" <EOL> coord . attributes [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> coord . attributes [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> coord . attributes [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> doc . appendChild ( coord ) <EOL> action = doc . createElement ( '<STR_LIT:action>' ) <EOL> workflow = doc . createElement ( '<STR_LIT>' ) <EOL> coord . appendChild ( action ) <EOL> action . appendChild ( workflow ) <EOL> x . add_text_element_to_tag ( doc , "<STR_LIT>" , '<STR_LIT>' , <EOL> "<STR_LIT>" ) <EOL> configuration = doc . createElement ( '<STR_LIT>' ) <EOL> workflow . appendChild ( configuration ) <EOL> default_configs = [ ] <EOL> if config_filter is not None : <EOL> default_configs = [ cfg [ '<STR_LIT:name>' ] for cfg in config_filter ] <EOL> for name in sorted ( coord_configs ) : <EOL> if name in default_configs or config_filter is None : <EOL> x . add_property_to_configuration ( doc , name , coord_configs [ name ] ) <EOL> return doc . toprettyxml ( indent = "<STR_LIT:U+0020>" ) <EOL> def _add_postfix ( self , constructed_dir ) : <EOL> def _append_slash_if_needed ( path ) : <EOL> if path [ - <NUM_LIT:1> ] != '<STR_LIT:/>' : <EOL> path += '<STR_LIT:/>' <EOL> return path <EOL> constructed_dir = _append_slash_if_needed ( constructed_dir ) <EOL> if CONF . job_workflow_postfix : <EOL> constructed_dir = '<STR_LIT>' . join ( [ str ( constructed_dir ) , <EOL> str ( CONF . job_workflow_postfix ) ] ) <EOL> return _append_slash_if_needed ( constructed_dir ) <EOL> def _resolve_external_hdfs_urls ( self , job_configs ) : <EOL> external_hdfs_urls = [ ] <EOL> for k , v in six . iteritems ( job_configs . get ( '<STR_LIT>' , { } ) ) : <EOL> if isinstance ( v , six . string_types ) and v . startswith ( "<STR_LIT>" ) : <EOL> external_hdfs_urls . append ( v ) <EOL> for k , v in six . iteritems ( job_configs . get ( '<STR_LIT>' , { } ) ) : <EOL> if isinstance ( v , six . string_types ) and v . startswith ( "<STR_LIT>" ) : <EOL> external_hdfs_urls . append ( v ) <EOL> for v in job_configs . get ( '<STR_LIT:args>' , [ ] ) : <EOL> if isinstance ( v , six . string_types ) and v . startswith ( "<STR_LIT>" ) : <EOL> external_hdfs_urls . append ( v ) <EOL> return external_hdfs_urls <EOL> def suspend_job ( self , job_execution ) : <EOL> return self . _manage_job ( job_execution , edp . JOB_ACTION_SUSPEND ) <EOL> def _manage_job ( self , job_execution , action ) : <EOL> if job_execution . oozie_job_id is not None : <EOL> client = self . _get_client ( ) <EOL> if action == edp . JOB_ACTION_SUSPEND : <EOL> client . suspend_job ( job_execution ) <EOL> return client . get_job_status ( job_execution ) </s>
