<s> from desktop . lib import thrift_util <EOL> from desktop . lib . conf import validate_port <EOL> from desktop . lib . exceptions import StructuredException <EOL> from desktop . lib . thrift_util import fixup_enums <EOL> from hadoop . api . jobtracker import Jobtracker <EOL> from hadoop . api . jobtracker . ttypes import ThriftJobID , ThriftTaskAttemptID , ThriftTaskType , ThriftTaskPhase , ThriftTaskID , ThriftTaskState , ThriftJobState , ThriftJobPriority , TaskNotFoundException , JobTrackerState , JobNotFoundException , ThriftTaskQueryState <EOL> from hadoop . api . common . ttypes import RequestContext <EOL> from thrift . transport import TTransport <EOL> import threading <EOL> VALID_TASK_STATES = set ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> VALID_TASK_TYPES = set ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> JT_THRIFT_TIMEOUT = <NUM_LIT:15> <EOL> DEFAULT_USER = "<STR_LIT>" <EOL> def test_jt_configuration ( cluster ) : <EOL> """<STR_LIT>""" <EOL> err = validate_port ( cluster . JT_THRIFT_PORT ) <EOL> if err : <EOL> return err <EOL> try : <EOL> jt = LiveJobTracker . from_conf ( cluster ) <EOL> jt . runtime_info ( ) <EOL> except TTransport . TTransportException : <EOL> msg = '<STR_LIT>' % ( cluster . HOST . get ( ) , cluster . JT_THRIFT_PORT . get ( ) ) <EOL> return [ ( cluster , msg ) ] <EOL> return [ ] <EOL> class LiveJobTracker ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , host , <EOL> thrift_port , <EOL> logical_name = None , <EOL> security_enabled = False , <EOL> kerberos_principal = "<STR_LIT>" ) : <EOL> self . client = thrift_util . get_client ( <EOL> Jobtracker . Client , host , thrift_port , <EOL> service_name = "<STR_LIT>" , <EOL> use_sasl = security_enabled , <EOL> kerberos_principal = kerberos_principal , <EOL> timeout_seconds = JT_THRIFT_TIMEOUT ) <EOL> self . host = host <EOL> self . thrift_port = thrift_port <EOL> self . logical_name = logical_name <EOL> self . security_enabled = security_enabled <EOL> self . thread_local = threading . local ( ) <EOL> self . setuser ( DEFAULT_USER ) <EOL> @ classmethod <EOL> def from_conf ( cls , conf ) : <EOL> return cls ( <EOL> conf . HOST . get ( ) , <EOL> conf . JT_THRIFT_PORT . get ( ) , <EOL> conf . LOGICAL_NAME . get ( ) , <EOL> security_enabled = conf . SECURITY_ENABLED . get ( ) , <EOL> kerberos_principal = conf . JT_KERBEROS_PRINCIPAL . get ( ) ) <EOL> def thriftjobid_from_string ( self , jobid ) : <EOL> """<STR_LIT>""" <EOL> _ , tid , jid = jobid . split ( "<STR_LIT:_>" ) <EOL> return ThriftJobID ( tid , int ( jid ) , jobid ) <EOL> def thriftattemptid_from_string ( self , attemptid ) : <EOL> """<STR_LIT>""" <EOL> _ , meat = attemptid . split ( '<STR_LIT:_>' , maxsplit = <NUM_LIT:1> ) <EOL> taskid , aid = meat . rsplit ( '<STR_LIT:_>' , maxsplit = <NUM_LIT:1> ) <EOL> return ThriftTaskAttemptID ( self . thrifttaskid_from_string ( taskid ) , int ( aid ) ) <EOL> def thrifttaskid_from_string ( self , taskid_str ) : <EOL> """<STR_LIT>""" <EOL> TASK_TYPE_MAP = { "<STR_LIT:m>" : ThriftTaskType . MAP , <EOL> "<STR_LIT:r>" : ThriftTaskType . REDUCE } <EOL> _ , tracker , jid , tasktype , tid = taskid_str . split ( '<STR_LIT:_>' ) <EOL> return ThriftTaskID ( ThriftJobID ( tracker , int ( jid ) ) , <EOL> TASK_TYPE_MAP [ tasktype ] , <EOL> int ( tid ) ) <EOL> def _fixup_tasktracker ( self , tracker ) : <EOL> """<STR_LIT>""" <EOL> if not tracker . taskReports : <EOL> return tracker <EOL> for tts in tracker . taskReports : <EOL> fixup_enums ( tts . taskID . taskID , { "<STR_LIT>" : ThriftTaskType } ) <EOL> self . _fixup_taskstatus ( tts ) <EOL> def _fixup_taskstatus ( self , taskstatus ) : <EOL> """<STR_LIT>""" <EOL> fixup_enums ( taskstatus , { "<STR_LIT:state>" : ThriftTaskState , "<STR_LIT>" : ThriftTaskPhase } ) <EOL> fixup_enums ( taskstatus . taskID . taskID , { "<STR_LIT>" : ThriftTaskType } ) <EOL> def _fixup_task_state ( self , tip ) : <EOL> """<STR_LIT>""" <EOL> if tip . complete : <EOL> tip . state = "<STR_LIT>" <EOL> elif tip . failed : <EOL> if not tip . execStartTime : <EOL> tip . state = "<STR_LIT>" <EOL> else : <EOL> tip . state = "<STR_LIT>" <EOL> elif tip . execStartTime : <EOL> tip . state = "<STR_LIT>" <EOL> else : <EOL> tip . state = "<STR_LIT>" <EOL> global VALID_TASK_STATES <EOL> assert tip . state in VALID_TASK_STATES <EOL> def _fixup_job ( self , job ) : <EOL> """<STR_LIT>""" <EOL> fixup_enums ( job . status , { "<STR_LIT>" : ThriftJobState , "<STR_LIT>" : ThriftJobPriority } ) <EOL> fixup_enums ( job , { "<STR_LIT>" : ThriftJobPriority } ) <EOL> if job . tasks is not None : <EOL> for tip in job . tasks . tasks : <EOL> fixup_enums ( tip . taskID , { "<STR_LIT>" : ThriftTaskType } ) <EOL> self . _fixup_task_state ( tip ) <EOL> for taskstatus in tip . taskStatuses . values ( ) : <EOL> self . _fixup_taskstatus ( taskstatus ) <EOL> def _fixup_task_in_progress ( self , tip ) : <EOL> fixup_enums ( tip . taskID , { "<STR_LIT>" : ThriftTaskType } ) <EOL> self . _fixup_task_state ( tip ) <EOL> for taskstatus in tip . taskStatuses . values ( ) : <EOL> self . _fixup_taskstatus ( taskstatus ) <EOL> def _fixup_retired_job ( self , job ) : <EOL> job . is_retired = True <EOL> def setuser ( self , user ) : <EOL> self . thread_local . request_context = RequestContext ( ) <EOL> if not self . thread_local . request_context . confOptions : <EOL> self . thread_local . request_context . confOptions = { } <EOL> self . thread_local . request_context . confOptions [ '<STR_LIT>' ] = user <EOL> self . thread_local . user = user <EOL> @ property <EOL> def user ( self ) : <EOL> return self . thread_local . user <EOL> @ property <EOL> def request_context ( self ) : <EOL> return self . thread_local . request_context <EOL> def queues ( self ) : <EOL> """<STR_LIT>""" <EOL> qs = self . client . getQueues ( self . thread_local . request_context ) <EOL> return qs <EOL> def cluster_status ( self ) : <EOL> """<STR_LIT>""" <EOL> cs = self . client . getClusterStatus ( self . thread_local . request_context ) <EOL> fixup_enums ( cs , { "<STR_LIT:state>" : JobTrackerState } ) <EOL> return cs <EOL> def runtime_info ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . client . getRuntimeInfo ( self . thread_local . request_context ) <EOL> def all_task_trackers ( self ) : <EOL> """<STR_LIT>""" <EOL> tts = self . client . getAllTrackers ( self . thread_local . request_context ) <EOL> for tracker in tts . trackers : <EOL> self . _fixup_tasktracker ( tracker ) <EOL> return tts <EOL> def active_trackers ( self ) : <EOL> """<STR_LIT>""" <EOL> tts = self . client . getActiveTrackers ( self . thread_local . request_context ) <EOL> for tracker in tts . trackers : <EOL> self . _fixup_tasktracker ( tracker ) <EOL> return tts <EOL> def blacklisted_trackers ( self ) : <EOL> """<STR_LIT>""" <EOL> tts = self . client . getBlacklistedTrackers ( self . thread_local . request_context ) <EOL> for tracker in tts . trackers : <EOL> self . _fixup_tasktracker ( tracker ) <EOL> return tts <EOL> def task_tracker ( self , name ) : <EOL> """<STR_LIT>""" <EOL> tracker = self . client . getTracker ( self . thread_local . request_context , name ) <EOL> if not tracker : <EOL> return None <EOL> self . _fixup_tasktracker ( tracker ) <EOL> return tracker <EOL> def get_job ( self , jobid ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> job = self . client . getJob ( self . thread_local . request_context , jobid ) <EOL> except JobNotFoundException , e : <EOL> raise StructuredException ( code = "<STR_LIT>" , message = "<STR_LIT>" % jobid . asString , data = jobid ) <EOL> self . _fixup_job ( job ) <EOL> return job <EOL> def running_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> joblist = self . client . getRunningJobs ( self . thread_local . request_context ) <EOL> for job in joblist . jobs : <EOL> self . _fixup_job ( job ) <EOL> return joblist <EOL> def completed_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> joblist = self . client . getCompletedJobs ( self . thread_local . request_context ) <EOL> for job in joblist . jobs : <EOL> self . _fixup_job ( job ) <EOL> return joblist <EOL> def get_retired_job ( self , jobid ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> job = self . client . getRetiredJob ( self . thread_local . request_context , jobid ) <EOL> except JobNotFoundException , e : <EOL> raise StructuredException ( code = "<STR_LIT>" , message = "<STR_LIT>" % jobid . asString , data = jobid ) <EOL> self . _fixup_job ( job ) <EOL> self . _fixup_retired_job ( job ) <EOL> return job <EOL> def retired_jobs ( self , status = None ) : <EOL> """<STR_LIT>""" <EOL> joblist = self . client . getRetiredJobs ( self . thread_local . request_context , status ) <EOL> for job in joblist . jobs : <EOL> self . _fixup_job ( job ) <EOL> self . _fixup_retired_job ( job ) <EOL> return joblist <EOL> def failed_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> joblist = self . client . getFailedJobs ( self . thread_local . request_context ) <EOL> for job in joblist . jobs : <EOL> self . _fixup_job ( job ) <EOL> return joblist <EOL> def killed_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> joblist = self . client . getKilledJobs ( self . thread_local . request_context ) <EOL> for job in joblist . jobs : <EOL> self . _fixup_job ( job ) <EOL> return joblist <EOL> def all_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> joblist = self . client . getAllJobs ( self . thread_local . request_context ) <EOL> for job in joblist . jobs : <EOL> self . _fixup_job ( job ) <EOL> return joblist <EOL> def get_job_count_by_user ( self , user ) : <EOL> """<STR_LIT>""" <EOL> return self . client . getUserJobCounts ( self . thread_local . request_context , user ) <EOL> def get_job_counters ( self , jobid ) : <EOL> """<STR_LIT>""" <EOL> return self . client . getJobCounters ( self . thread_local . request_context , jobid ) <EOL> def get_job_counter_rollups ( self , jobid ) : <EOL> """<STR_LIT>""" <EOL> return self . client . getJobCounterRollups ( self . thread_local . request_context , jobid ) <EOL> def get_task_list ( self , jobid , task_types , task_states , task_text , count , offset ) : <EOL> """<STR_LIT>""" <EOL> ttask_types = [ ThriftTaskType . _NAMES_TO_VALUES [ x . upper ( ) ] for x in task_types ] <EOL> ttask_states = [ ThriftTaskQueryState . _NAMES_TO_VALUES [ x . upper ( ) ] for x in task_states ] <EOL> tip_list = self . client . getTaskList ( <EOL> self . thread_local . request_context , jobid , ttask_types , ttask_states , task_text , count , offset ) <EOL> for tip in tip_list . tasks : <EOL> self . _fixup_task_in_progress ( tip ) <EOL> return tip_list <EOL> def get_task ( self , jobid , taskid ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> tip = self . client . getTask ( self . thread_local . request_context , taskid ) <EOL> except JobNotFoundException , e : <EOL> raise StructuredException ( code = "<STR_LIT>" , message = "<STR_LIT>" % jobid . asString , data = jobid ) <EOL> except TaskNotFoundException , e : <EOL> raise StructuredException ( code = "<STR_LIT>" , message = "<STR_LIT>" % taskid . asString , data = taskid ) <EOL> self . _fixup_task_in_progress ( tip ) <EOL> return tip <EOL> def get_current_time ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . client . getCurrentTime ( self . thread_local . request_context ) <EOL> def get_job_xml ( self , jobid ) : <EOL> """<STR_LIT>""" <EOL> return self . client . getJobConfXML ( self . thread_local . request_context , jobid ) <EOL> def kill_job ( self , jobid ) : <EOL> """<STR_LIT>""" <EOL> return self . client . killJob ( self . thread_local . request_context , jobid ) <EOL> def kill_task_attempt ( self , attemptid ) : <EOL> """<STR_LIT>""" <EOL> return self . client . killTaskAttempt ( self . thread_local . request_context , attemptid ) <EOL> def set_job_priority ( self , jobid , priority ) : <EOL> """<STR_LIT>""" <EOL> return self . client . setJobPriority ( self . thread_local . request_context , jobid , priority ) <EOL> def get_delegation_token ( self , principal ) : <EOL> return self . client . getDelegationToken ( self . thread_local . request_context , principal ) . delegationTokenBytes </s>
