<s> import logging <EOL> from desktop . lib . exceptions_renderable import PopupException <EOL> from desktop . lib . paginator import Paginator <EOL> from desktop . lib . rest . http_client import RestException <EOL> from hadoop import cluster <EOL> from hadoop . cluster import jt_ha , rm_ha <EOL> from hadoop . api . jobtracker . ttypes import ThriftJobPriority , TaskTrackerNotFoundException , ThriftJobState <EOL> import hadoop . yarn . history_server_api as history_server_api <EOL> import hadoop . yarn . mapreduce_api as mapreduce_api <EOL> import hadoop . yarn . node_manager_api as node_manager_api <EOL> import hadoop . yarn . resource_manager_api as resource_manager_api <EOL> import hadoop . yarn . spark_history_server_api as spark_history_server_api <EOL> from jobbrowser . conf import SHARE_JOBS <EOL> from jobbrowser . models import Job , JobLinkage , TaskList , Tracker <EOL> from jobbrowser . yarn_models import Application , Job as YarnJob , KilledJob as KilledYarnJob , Container , SparkJob <EOL> LOG = logging . getLogger ( __name__ ) <EOL> _DEFAULT_OBJ_PER_PAGINATION = <NUM_LIT:10> <EOL> def get_api ( user , jt ) : <EOL> if cluster . is_yarn ( ) : <EOL> return YarnApi ( user ) <EOL> else : <EOL> return JtApi ( jt ) <EOL> class JobBrowserApi ( object ) : <EOL> def paginate_task ( self , task_list , pagenum ) : <EOL> paginator = Paginator ( task_list , _DEFAULT_OBJ_PER_PAGINATION ) <EOL> return paginator . page ( pagenum ) <EOL> class JtApi ( JobBrowserApi ) : <EOL> def __init__ ( self , jt ) : <EOL> self . jt = jt <EOL> @ jt_ha <EOL> def get_job_link ( self , jobid ) : <EOL> return JobLinkage ( self . jt , jobid ) <EOL> @ jt_ha <EOL> def get_job ( self , jobid ) : <EOL> return Job . from_id ( jt = self . jt , jobid = jobid ) <EOL> @ jt_ha <EOL> def get_jobs ( self , user , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> jobfunc = { <EOL> "<STR_LIT>" : ( self . jt . completed_jobs , ThriftJobState . SUCCEEDED ) , <EOL> "<STR_LIT>" : ( self . jt . completed_jobs , ThriftJobState . SUCCEEDED ) , <EOL> "<STR_LIT>" : ( self . jt . running_jobs , ThriftJobState . RUNNING ) , <EOL> "<STR_LIT>" : ( self . jt . failed_jobs , ThriftJobState . FAILED ) , <EOL> "<STR_LIT>" : ( self . jt . killed_jobs , ThriftJobState . KILLED ) , <EOL> "<STR_LIT:all>" : ( self . jt . all_jobs , None ) , <EOL> None : ( self . jt . all_jobs , None ) <EOL> } <EOL> selection = kwargs . pop ( '<STR_LIT:state>' ) <EOL> retired = kwargs . pop ( '<STR_LIT>' ) <EOL> jobs = jobfunc [ selection ] [ <NUM_LIT:0> ] ( ) . jobs <EOL> if retired : <EOL> jobs += self . jt . retired_jobs ( jobfunc [ selection ] [ <NUM_LIT:1> ] ) . jobs <EOL> return self . filter_jobs ( user , jobs , ** kwargs ) <EOL> @ jt_ha <EOL> def filter_jobs ( self , user , jobs , ** kwargs ) : <EOL> check_permission = not SHARE_JOBS . get ( ) and not user . is_superuser <EOL> limit = kwargs . pop ( '<STR_LIT>' , <NUM_LIT> ) <EOL> return [ Job . from_thriftjob ( self . jt , j ) <EOL> for j in self . _filter_jobs ( jobs , ** kwargs ) <EOL> if not check_permission or user . is_superuser or j . profile . user == user . username ] [ : limit ] <EOL> def _filter_jobs ( self , jobs , username = None , text = None ) : <EOL> def predicate ( job ) : <EOL> """<STR_LIT>""" <EOL> if username and username not in job . profile . user : <EOL> return False <EOL> if text : <EOL> search = text . lower ( ) <EOL> saw_text = False <EOL> for t in [ job . profile . user , <EOL> job . profile . name , <EOL> job . jobID . asString , <EOL> job . profile . queueName , <EOL> job . priorityAsString <EOL> ] : <EOL> if search in t . lower ( ) : <EOL> saw_text = True <EOL> break <EOL> if not saw_text : <EOL> return False <EOL> return True <EOL> return filter ( predicate , jobs ) <EOL> @ jt_ha <EOL> def get_tasks ( self , jobid , ** filters ) : <EOL> return TaskList . select ( self . jt , <EOL> jobid , <EOL> filters [ '<STR_LIT>' ] , <EOL> filters [ '<STR_LIT>' ] , <EOL> filters [ '<STR_LIT>' ] , <EOL> _DEFAULT_OBJ_PER_PAGINATION , <EOL> _DEFAULT_OBJ_PER_PAGINATION * ( filters [ '<STR_LIT>' ] - <NUM_LIT:1> ) ) <EOL> @ jt_ha <EOL> def get_tracker ( self , trackerid ) : <EOL> return Tracker . from_name ( self . jt , trackerid ) <EOL> class YarnApi ( JobBrowserApi ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , user ) : <EOL> self . user = user <EOL> self . resource_manager_api = resource_manager_api . get_resource_manager ( user . username ) <EOL> self . mapreduce_api = mapreduce_api . get_mapreduce_api ( user . username ) <EOL> self . history_server_api = history_server_api . get_history_server_api ( user . username ) <EOL> self . spark_history_server_api = spark_history_server_api . get_history_server_api ( ) <EOL> def get_job_link ( self , job_id ) : <EOL> return self . get_job ( job_id ) <EOL> @ rm_ha <EOL> def get_jobs ( self , user , ** kwargs ) : <EOL> state_filters = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , } <EOL> filters = { } <EOL> if kwargs [ '<STR_LIT:username>' ] : <EOL> filters [ '<STR_LIT:user>' ] = kwargs [ '<STR_LIT:username>' ] <EOL> if kwargs [ '<STR_LIT:state>' ] and kwargs [ '<STR_LIT:state>' ] != '<STR_LIT:all>' : <EOL> filters [ '<STR_LIT>' ] = state_filters [ kwargs [ '<STR_LIT:state>' ] ] <EOL> if kwargs . get ( '<STR_LIT>' ) : <EOL> filters [ '<STR_LIT>' ] = kwargs [ '<STR_LIT>' ] <EOL> json = self . resource_manager_api . apps ( ** filters ) <EOL> if type ( json ) == str and '<STR_LIT>' in json : <EOL> raise Exception ( json ) <EOL> if json [ '<STR_LIT>' ] : <EOL> jobs = [ Application ( app ) for app in json [ '<STR_LIT>' ] [ '<STR_LIT>' ] ] <EOL> else : <EOL> return [ ] <EOL> if kwargs [ '<STR_LIT:text>' ] : <EOL> text = kwargs [ '<STR_LIT:text>' ] . lower ( ) <EOL> jobs = filter ( lambda job : <EOL> text in job . name . lower ( ) or <EOL> text in job . id . lower ( ) or <EOL> text in job . user . lower ( ) or <EOL> text in job . queue . lower ( ) , jobs ) <EOL> return self . filter_jobs ( user , jobs ) <EOL> def filter_jobs ( self , user , jobs , ** kwargs ) : <EOL> check_permission = not SHARE_JOBS . get ( ) and not user . is_superuser <EOL> return filter ( lambda job : <EOL> not check_permission or <EOL> user . is_superuser or <EOL> job . user == user . username , jobs ) <EOL> @ rm_ha <EOL> def get_job ( self , jobid ) : <EOL> job_id = jobid . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> app_id = jobid . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> app = self . resource_manager_api . app ( app_id ) [ '<STR_LIT>' ] <EOL> if app [ '<STR_LIT>' ] in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if app [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> job = SparkJob ( app , rm_api = self . resource_manager_api , hs_api = self . spark_history_server_api ) <EOL> elif app [ '<STR_LIT:state>' ] == '<STR_LIT>' : <EOL> job = KilledYarnJob ( self . resource_manager_api , app ) <EOL> else : <EOL> resp = self . history_server_api . job ( self . user , job_id ) <EOL> job = YarnJob ( self . history_server_api , resp [ '<STR_LIT>' ] ) <EOL> else : <EOL> if app [ '<STR_LIT:state>' ] == '<STR_LIT>' : <EOL> raise ApplicationNotRunning ( app_id , app ) <EOL> elif app [ '<STR_LIT:state>' ] in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) and app [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> resp = self . mapreduce_api . job ( self . user , job_id ) <EOL> job = YarnJob ( self . mapreduce_api , resp [ '<STR_LIT>' ] ) <EOL> else : <EOL> job = Application ( app , self . resource_manager_api ) <EOL> except RestException , e : <EOL> if e . code == <NUM_LIT> : <EOL> resp = self . history_server_api . job ( self . user , job_id ) <EOL> job = YarnJob ( self . history_server_api , resp [ '<STR_LIT>' ] ) <EOL> else : <EOL> raise JobExpired ( app_id ) <EOL> except ApplicationNotRunning , e : <EOL> raise e <EOL> except Exception , e : <EOL> raise PopupException ( '<STR_LIT>' % ( jobid , e ) , detail = e ) <EOL> return job <EOL> def get_tasks ( self , jobid , ** filters ) : <EOL> filters . pop ( '<STR_LIT>' ) <EOL> return self . get_job ( jobid ) . filter_tasks ( ** filters ) <EOL> def get_task ( self , jobid , task_id ) : <EOL> return self . get_job ( jobid ) . task ( task_id ) <EOL> def get_tracker ( self , node_manager_http_address , container_id ) : <EOL> api = node_manager_api . get_node_manager_api ( '<STR_LIT>' + node_manager_http_address ) <EOL> return Container ( api . container ( container_id ) ) <EOL> class ApplicationNotRunning ( Exception ) : <EOL> def __init__ ( self , application_id , job ) : <EOL> self . application_id = application_id <EOL> self . job = job <EOL> class JobExpired ( Exception ) : <EOL> def __init__ ( self , job ) : <EOL> super ( JobExpired , self ) . __init__ ( '<STR_LIT>' % job ) <EOL> self . job = job </s>
