<s> """<STR_LIT>""" <EOL> import posixpath <EOL> import re <EOL> from mrjob . util import file_ext <EOL> from . ids import _add_implied_task_id <EOL> from . ids import _to_job_id <EOL> from . log4j import _parse_hadoop_log4j_records <EOL> from . wrap import _cat_log <EOL> from . wrap import _ls_logs <EOL> _JAVA_TRACEBACK_RE = re . compile ( <EOL> r'<STR_LIT>' , <EOL> re . MULTILINE ) <EOL> _OPENING_FOR_READING_RE = re . compile ( <EOL> r"<STR_LIT>" ) <EOL> _PRE_YARN_TASK_SYSLOG_PATH_RE = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' ) <EOL> _TASK_STDERR_IGNORE_RE = re . compile ( r'<STR_LIT>' ) <EOL> _YARN_INPUT_SPLIT_RE = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' ) <EOL> _YARN_TASK_SYSLOG_PATH_RE = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' ) <EOL> def _ls_task_syslogs ( fs , log_dir_stream , application_id = None , job_id = None ) : <EOL> """<STR_LIT>""" <EOL> return _ls_logs ( fs , log_dir_stream , _match_task_syslog_path , <EOL> application_id = application_id , <EOL> job_id = job_id ) <EOL> def _match_task_syslog_path ( path , application_id = None , job_id = None ) : <EOL> """<STR_LIT>""" <EOL> m = _PRE_YARN_TASK_SYSLOG_PATH_RE . match ( path ) <EOL> if m : <EOL> if job_id and job_id != _to_job_id ( m . group ( '<STR_LIT>' ) ) : <EOL> return None <EOL> return dict ( <EOL> attempt_id = m . group ( '<STR_LIT>' ) ) <EOL> m = _YARN_TASK_SYSLOG_PATH_RE . match ( path ) <EOL> if m : <EOL> if application_id and application_id != m . group ( '<STR_LIT>' ) : <EOL> return None <EOL> return dict ( <EOL> application_id = m . group ( '<STR_LIT>' ) , <EOL> container_id = m . group ( '<STR_LIT>' ) ) <EOL> return None <EOL> def _interpret_task_logs ( fs , matches , partial = True , stderr_callback = None ) : <EOL> """<STR_LIT>""" <EOL> result = { } <EOL> for match in matches : <EOL> syslog_path = match [ '<STR_LIT:path>' ] <EOL> error = _parse_task_syslog ( _cat_log ( fs , syslog_path ) ) <EOL> if not error . get ( '<STR_LIT>' ) : <EOL> continue <EOL> error [ '<STR_LIT>' ] [ '<STR_LIT:path>' ] = syslog_path <EOL> for id_key in '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' : <EOL> if id_key in match : <EOL> error [ id_key ] = match [ id_key ] <EOL> _add_implied_task_id ( error ) <EOL> stderr_path = _syslog_to_stderr_path ( syslog_path ) <EOL> if fs . exists ( stderr_path ) : <EOL> if stderr_callback : <EOL> stderr_callback ( stderr_path ) <EOL> task_error = _parse_task_stderr ( _cat_log ( fs , stderr_path ) ) <EOL> if task_error : <EOL> task_error [ '<STR_LIT:path>' ] = stderr_path <EOL> error [ '<STR_LIT>' ] = task_error <EOL> result . setdefault ( '<STR_LIT>' , [ ] ) <EOL> result [ '<STR_LIT>' ] . append ( error ) <EOL> if partial : <EOL> result [ '<STR_LIT>' ] = True <EOL> break <EOL> return result <EOL> def _syslog_to_stderr_path ( path ) : <EOL> """<STR_LIT>""" <EOL> stem , filename = posixpath . split ( path ) <EOL> return posixpath . join ( stem , '<STR_LIT>' + file_ext ( filename ) ) <EOL> def _parse_task_syslog ( lines ) : <EOL> """<STR_LIT>""" <EOL> result = { } <EOL> for record in _parse_hadoop_log4j_records ( lines ) : <EOL> message = record [ '<STR_LIT:message>' ] <EOL> m = _OPENING_FOR_READING_RE . match ( message ) <EOL> if m : <EOL> result [ '<STR_LIT>' ] = dict ( path = m . group ( '<STR_LIT:path>' ) ) <EOL> continue <EOL> m = _YARN_INPUT_SPLIT_RE . match ( message ) <EOL> if m : <EOL> result [ '<STR_LIT>' ] = dict ( <EOL> path = m . group ( '<STR_LIT:path>' ) , <EOL> start_line = int ( m . group ( '<STR_LIT>' ) ) , <EOL> num_lines = int ( m . group ( '<STR_LIT>' ) ) ) <EOL> continue <EOL> m = _JAVA_TRACEBACK_RE . search ( message ) <EOL> if m : <EOL> result [ '<STR_LIT>' ] = dict ( <EOL> message = message , <EOL> num_lines = record [ '<STR_LIT>' ] , <EOL> start_line = record [ '<STR_LIT>' ] , <EOL> ) <EOL> break <EOL> return result <EOL> def _parse_task_stderr ( lines ) : <EOL> """<STR_LIT>""" <EOL> task_error = None <EOL> for line_num , line in enumerate ( lines ) : <EOL> line = line . rstrip ( '<STR_LIT:\r\n>' ) <EOL> if _TASK_STDERR_IGNORE_RE . match ( line ) : <EOL> if task_error and '<STR_LIT>' not in task_error : <EOL> task_error [ '<STR_LIT>' ] = line_num - task_error [ '<STR_LIT>' ] <EOL> continue <EOL> elif not task_error or line . startswith ( '<STR_LIT>' ) : <EOL> task_error = dict ( <EOL> message = line , <EOL> start_line = line_num ) <EOL> else : <EOL> task_error [ '<STR_LIT:message>' ] += '<STR_LIT:\n>' + line <EOL> if task_error : <EOL> if '<STR_LIT>' not in task_error : <EOL> task_error [ '<STR_LIT>' ] = line_num + <NUM_LIT:1> - task_error [ '<STR_LIT>' ] <EOL> return task_error <EOL> else : <EOL> return None </s>
