<s> import datetime <EOL> from oslo_config import cfg <EOL> from oslo_log import log <EOL> from oslo_utils import timeutils <EOL> from sahara import conductor as c <EOL> from sahara import context <EOL> from sahara import exceptions as e <EOL> from sahara . i18n import _ <EOL> from sahara . i18n import _LE <EOL> from sahara . i18n import _LI <EOL> from sahara . i18n import _LW <EOL> from sahara . service . edp import job_utils <EOL> from sahara . service . edp . oozie import engine as oozie_engine <EOL> from sahara . service . edp . spark import engine as spark_engine <EOL> from sahara . service . edp . storm import engine as storm_engine <EOL> from sahara . utils import cluster as c_u <EOL> from sahara . utils import edp <EOL> from sahara . utils import proxy as p <EOL> LOG = log . getLogger ( __name__ ) <EOL> CONF = cfg . CONF <EOL> conductor = c . API <EOL> ENGINES = [ oozie_engine . OozieJobEngine , <EOL> spark_engine . SparkJobEngine , <EOL> storm_engine . StormJobEngine ] <EOL> def _get_job_type ( job_execution ) : <EOL> return conductor . job_get ( context . ctx ( ) , job_execution . job_id ) . type <EOL> def get_job_engine ( cluster , job_execution ) : <EOL> return job_utils . get_plugin ( cluster ) . get_edp_engine ( cluster , <EOL> _get_job_type ( <EOL> job_execution ) ) <EOL> def _write_job_status ( job_execution , job_info ) : <EOL> update = { "<STR_LIT:info>" : job_info } <EOL> if job_info [ '<STR_LIT:status>' ] in edp . JOB_STATUSES_TERMINATED : <EOL> update [ '<STR_LIT>' ] = datetime . datetime . now ( ) <EOL> job_configs = p . delete_proxy_user_for_job_execution ( job_execution ) <EOL> if job_configs : <EOL> update [ '<STR_LIT>' ] = job_configs <EOL> return conductor . job_execution_update ( context . ctx ( ) , <EOL> job_execution , <EOL> update ) <EOL> def _update_job_status ( engine , job_execution ) : <EOL> job_info = engine . get_job_status ( job_execution ) <EOL> if job_info is not None : <EOL> job_execution = _write_job_status ( job_execution , job_info ) <EOL> return job_execution <EOL> def _update_job_execution_extra ( cluster , job_execution ) : <EOL> if ( ( CONF . use_namespaces and not CONF . use_floating_ips ) or <EOL> CONF . proxy_command ) : <EOL> info = cluster . node_groups [ <NUM_LIT:0> ] . instances [ <NUM_LIT:0> ] . remote ( ) . get_neutron_info ( ) <EOL> extra = job_execution . extra . copy ( ) <EOL> extra [ '<STR_LIT>' ] = info <EOL> job_execution = conductor . job_execution_update ( <EOL> context . ctx ( ) , job_execution . id , { '<STR_LIT>' : extra } ) <EOL> return job_execution <EOL> def _run_job ( job_execution_id ) : <EOL> ctx = context . ctx ( ) <EOL> job_execution = conductor . job_execution_get ( ctx , job_execution_id ) <EOL> cluster = conductor . cluster_get ( ctx , job_execution . cluster_id ) <EOL> if cluster is None or cluster . status != c_u . CLUSTER_STATUS_ACTIVE : <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> return <EOL> eng = get_job_engine ( cluster , job_execution ) <EOL> if eng is None : <EOL> raise e . EDPError ( _ ( "<STR_LIT>" ) <EOL> % _get_job_type ( job_execution ) ) <EOL> job_execution = _update_job_execution_extra ( cluster , job_execution ) <EOL> if job_execution . job_configs . job_execution_info . get ( '<STR_LIT>' <EOL> ) == '<STR_LIT>' : <EOL> jid , status , extra = eng . run_scheduled_job ( job_execution ) <EOL> else : <EOL> jid , status , extra = eng . run_job ( job_execution ) <EOL> update_dict = { '<STR_LIT>' : jid , <EOL> '<STR_LIT>' : datetime . datetime . now ( ) } <EOL> if status : <EOL> update_dict [ '<STR_LIT:info>' ] = { '<STR_LIT:status>' : status } <EOL> if extra : <EOL> curr_extra = job_execution . extra . copy ( ) <EOL> if '<STR_LIT>' in curr_extra : <EOL> curr_extra [ '<STR_LIT>' ] = curr_extra [ '<STR_LIT>' ] . copy ( ) <EOL> curr_extra . update ( extra ) <EOL> update_dict [ '<STR_LIT>' ] = curr_extra <EOL> job_execution = conductor . job_execution_update ( <EOL> ctx , job_execution , update_dict ) <EOL> def run_job ( job_execution_id ) : <EOL> try : <EOL> _run_job ( job_execution_id ) <EOL> except Exception as ex : <EOL> LOG . warning ( <EOL> _LW ( "<STR_LIT>" ) . format ( <EOL> reason = ex ) ) <EOL> job_execution = conductor . job_execution_get ( <EOL> context . ctx ( ) , job_execution_id ) <EOL> if job_execution . engine_job_id is not None : <EOL> cancel_job ( job_execution_id ) <EOL> conductor . job_execution_update ( <EOL> context . ctx ( ) , job_execution_id , <EOL> { '<STR_LIT:info>' : { '<STR_LIT:status>' : edp . JOB_STATUS_FAILED } , <EOL> '<STR_LIT>' : datetime . datetime . now ( ) , <EOL> '<STR_LIT>' : datetime . datetime . now ( ) } ) <EOL> def cancel_job ( job_execution_id ) : <EOL> ctx = context . ctx ( ) <EOL> job_execution = conductor . job_execution_get ( ctx , job_execution_id ) <EOL> if job_execution . info [ '<STR_LIT:status>' ] in edp . JOB_STATUSES_TERMINATED : <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> return job_execution <EOL> cluster = conductor . cluster_get ( ctx , job_execution . cluster_id ) <EOL> if cluster is None : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> return job_execution <EOL> engine = get_job_engine ( cluster , job_execution ) <EOL> if engine is not None : <EOL> job_execution = conductor . job_execution_update ( <EOL> ctx , job_execution_id , <EOL> { '<STR_LIT:info>' : { '<STR_LIT:status>' : edp . JOB_STATUS_TOBEKILLED } } ) <EOL> timeout = CONF . job_canceling_timeout <EOL> s_time = timeutils . utcnow ( ) <EOL> while timeutils . delta_seconds ( s_time , timeutils . utcnow ( ) ) < timeout : <EOL> if job_execution . info [ '<STR_LIT:status>' ] not in edp . JOB_STATUSES_TERMINATED : <EOL> try : <EOL> job_info = engine . cancel_job ( job_execution ) <EOL> except Exception as ex : <EOL> job_info = None <EOL> LOG . warning ( <EOL> _LW ( "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( error = ex ) ) <EOL> if job_info is not None : <EOL> job_execution = _write_job_status ( job_execution , job_info ) <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> return job_execution <EOL> context . sleep ( <NUM_LIT:3> ) <EOL> job_execution = conductor . job_execution_get ( <EOL> ctx , job_execution_id ) <EOL> if not job_execution : <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> return job_execution <EOL> else : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) . format ( <EOL> status = job_execution . info [ '<STR_LIT:status>' ] ) ) <EOL> return job_execution <EOL> else : <EOL> raise e . CancelingFailed ( _ ( '<STR_LIT>' ) <EOL> % job_execution . id ) <EOL> def get_job_status ( job_execution_id ) : <EOL> ctx = context . ctx ( ) <EOL> job_execution = conductor . job_execution_get ( ctx , job_execution_id ) <EOL> cluster = conductor . cluster_get ( ctx , job_execution . cluster_id ) <EOL> if ( cluster is not None and <EOL> cluster . status == c_u . CLUSTER_STATUS_ACTIVE ) : <EOL> engine = get_job_engine ( cluster , job_execution ) <EOL> if engine is not None : <EOL> job_execution = _update_job_status ( engine , <EOL> job_execution ) <EOL> return job_execution <EOL> def update_job_status ( job_execution_id ) : <EOL> try : <EOL> get_job_status ( job_execution_id ) <EOL> except Exception as e : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) <EOL> . format ( job = job_execution_id , error = e ) ) <EOL> def update_job_statuses ( cluster_id = None ) : <EOL> ctx = context . ctx ( ) <EOL> kwargs = { '<STR_LIT>' : None } <EOL> if cluster_id : <EOL> kwargs . update ( { '<STR_LIT>' : cluster_id } ) <EOL> for je in conductor . job_execution_get_all ( ctx , ** kwargs ) : <EOL> update_job_status ( je . id ) <EOL> def get_job_config_hints ( job_type ) : <EOL> for eng in ENGINES : <EOL> if job_type in eng . get_supported_job_types ( ) : <EOL> return eng . get_possible_job_config ( job_type ) <EOL> def suspend_job ( job_execution_id ) : <EOL> ctx = context . ctx ( ) <EOL> job_execution = conductor . job_execution_get ( ctx , job_execution_id ) <EOL> if job_execution . info [ '<STR_LIT:status>' ] not in edp . JOB_STATUSES_SUSPENDIBLE : <EOL> raise e . SuspendingFailed ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) . format ( <EOL> status = job_execution . info [ '<STR_LIT:status>' ] ) <EOL> cluster = conductor . cluster_get ( ctx , job_execution . cluster_id ) <EOL> engine = get_job_engine ( cluster , job_execution ) <EOL> job_execution = conductor . job_execution_update ( <EOL> ctx , job_execution_id , { <EOL> '<STR_LIT:info>' : { '<STR_LIT:status>' : edp . JOB_STATUS_TOBESUSPENDED } } ) <EOL> try : <EOL> job_info = engine . suspend_job ( job_execution ) <EOL> except Exception as ex : <EOL> job_info = None <EOL> conductor . job_execution_update ( <EOL> ctx , job_execution_id , { '<STR_LIT:info>' : { <EOL> '<STR_LIT:status>' : edp . JOB_STATUS_SUSPEND_FAILED } } ) <EOL> raise e . SuspendingFailed ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) . format ( error = ex ) <EOL> if job_info is not None : <EOL> job_execution = _write_job_status ( job_execution , job_info ) <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> return job_execution <EOL> conductor . job_execution_update ( <EOL> ctx , job_execution_id , { '<STR_LIT:info>' : { <EOL> '<STR_LIT:status>' : edp . JOB_STATUS_SUSPEND_FAILED } } ) <EOL> raise e . SuspendingFailed ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) . format ( jid = job_execution_id ) </s>
