#!/usr/bin/python
|
"""
|
Usage: ./cron_scripts/log_distiller.py job_id path_to_logfile
|
If the job_id is a suite it will find all subjobs.
|
You need to change the location of the log it will parse.
|
The job_id needs to be in the afe database.
|
"""
|
import abc
|
import datetime
|
import os
|
import re
|
import pprint
|
import subprocess
|
import sys
|
import time
|
|
import common
|
from autotest_lib.server import frontend
|
|
|
LOGFIE = './logs/scheduler.log.2014-04-17-16.51.47'
|
# logfile name format: scheduler.log.2014-02-14-18.10.56
|
time_format = '%Y-%m-%d-%H.%M.%S'
|
logfile_regex = r'scheduler.log.([0-9,.,-]+)'
|
logdir = os.path.join('/usr/local/autotest', 'logs')
|
|
class StateMachineViolation(Exception):
|
pass
|
|
|
class LogLineException(Exception):
|
pass
|
|
|
def should_process_log(time_str, time_format, cutoff_days=7):
|
"""Returns true if the logs was created after cutoff days.
|
|
@param time_str: A string representing the time.
|
eg: 2014-02-14-18.10.56
|
@param time_format: A string representing the format of the time string.
|
ref: http://docs.python.org/2/library/datetime.html#strftime-strptime-behavior
|
@param cutoff_days: Int representind the cutoff in days.
|
|
@return: Returns True if time_str has aged more than cutoff_days.
|
"""
|
log_time = datetime.datetime.strptime(time_str, time_format)
|
now = datetime.datetime.strptime(time.strftime(time_format), time_format)
|
cutoff = now - datetime.timedelta(days=cutoff_days)
|
return log_time < cutoff
|
|
|
def apply_regex(regex, line):
|
"""Simple regex applicator.
|
|
@param regex: Regex to apply.
|
@param line: The line to apply regex on.
|
|
@return: A tuple with the matching groups, if there was a match.
|
"""
|
log_match = re.match(regex, line)
|
if log_match:
|
return log_match.groups()
|
|
|
class StateMachineParser(object):
|
"""Abstract class that enforces state transition ordering.
|
|
Classes inheriting from StateMachineParser need to define an
|
expected_transitions dictionary. The SMP will pop 'to' states
|
from the dictionary as they occur, so you cannot same state transitions
|
unless you specify 2 of them.
|
"""
|
__metaclass__ = abc.ABCMeta
|
|
|
@abc.abstractmethod
|
def __init__(self):
|
self.visited_states = []
|
self.expected_transitions = {}
|
|
|
def advance_state(self, from_state, to_state):
|
"""Checks that a transition is valid.
|
|
@param from_state: A string representind the state the host is leaving.
|
@param to_state: The state The host is going to, represented as a string.
|
|
@raises LogLineException: If an invalid state transition was
|
detected.
|
"""
|
# TODO: Updating to the same state is a waste of bw.
|
if from_state and from_state == to_state:
|
return ('Updating to the same state is a waste of BW: %s->%s' %
|
(from_state, to_state))
|
return
|
|
if (from_state in self.expected_transitions and
|
to_state in self.expected_transitions[from_state]):
|
self.expected_transitions[from_state].remove(to_state)
|
self.visited_states.append(to_state)
|
else:
|
return (from_state, to_state)
|
|
|
class SingleJobHostSMP(StateMachineParser):
|
def __init__(self):
|
self.visited_states = []
|
self.expected_transitions = {
|
'Ready': ['Resetting', 'Verifying', 'Pending', 'Provisioning'],
|
'Resetting': ['Ready', 'Provisioning'],
|
'Pending': ['Running'],
|
'Provisioning': ['Repairing'],
|
'Running': ['Ready']
|
}
|
|
|
def check_transitions(self, hostline):
|
if hostline.line_info['field'] == 'status':
|
self.advance_state(hostline.line_info['state'],
|
hostline.line_info['value'])
|
|
|
class SingleJobHqeSMP(StateMachineParser):
|
def __init__(self):
|
self.visited_states = []
|
self.expected_transitions = {
|
'Queued': ['Starting', 'Resetting', 'Aborted'],
|
'Resetting': ['Pending', 'Provisioning'],
|
'Provisioning': ['Pending', 'Queued', 'Repairing'],
|
'Pending': ['Starting'],
|
'Starting': ['Running'],
|
'Running': ['Gathering', 'Parsing'],
|
'Gathering': ['Parsing'],
|
'Parsing': ['Completed', 'Aborted']
|
}
|
|
|
def check_transitions(self, hqeline):
|
invalid_states = self.advance_state(
|
hqeline.line_info['from_state'], hqeline.line_info['to_state'])
|
if not invalid_states:
|
return
|
|
# Deal with repair.
|
if (invalid_states[0] == 'Queued' and
|
'Running' in self.visited_states):
|
raise StateMachineViolation('Unrecognized state transition '
|
'%s->%s, expected transitions are %s' %
|
(invalid_states[0], invalid_states[1],
|
self.expected_transitions))
|
|
|
class LogLine(object):
|
"""Line objects.
|
|
All classes inheriting from LogLine represent a line of some sort.
|
A line is responsible for parsing itself, and invoking an SMP to
|
validate state transitions. A line can be part of several state machines.
|
"""
|
line_format = '%s'
|
|
|
def __init__(self, state_machine_parsers):
|
"""
|
@param state_machine_parsers: A list of smp objects to use to validate
|
state changes on these types of lines..
|
"""
|
self.smps = state_machine_parsers
|
|
# Because, this is easier to flush.
|
self.line_info = {}
|
|
|
def parse_line(self, line):
|
"""Apply a line regex and save any information the parsed line contains.
|
|
@param line: A string representing a line.
|
"""
|
# Regex for all the things.
|
line_rgx = '(.*)'
|
parsed_line = apply_regex(line_rgx, line)
|
if parsed_line:
|
self.line_info['line'] = parsed_line[0]
|
|
|
def flush(self):
|
"""Call any state machine parsers, persist line info if needed.
|
"""
|
for smp in self.smps:
|
smp.check_transitions(self)
|
# TODO: persist this?
|
self.line_info={}
|
|
|
def format_line(self):
|
try:
|
return self.line_format % self.line_info
|
except KeyError:
|
return self.line_info['line']
|
|
|
class TimeLine(LogLine):
|
"""Filters timestamps for scheduler logs.
|
"""
|
|
def parse_line(self, line):
|
super(TimeLine, self).parse_line(line)
|
|
# Regex for isolating the date and time from scheduler logs, eg:
|
# 02/16 16:04:36.573 INFO |scheduler_:0574|...
|
line_rgx = '([0-9,/,:,., ]+)(.*)'
|
parsed_line = apply_regex(line_rgx, self.line_info['line'])
|
if parsed_line:
|
self.line_info['time'] = parsed_line[0]
|
self.line_info['line'] = parsed_line[1]
|
|
|
class HostLine(TimeLine):
|
"""Manages hosts line parsing.
|
"""
|
line_format = (' \t\t %(time)s %(host)s, currently in %(state)s, '
|
'updated %(field)s->%(value)s')
|
|
|
def record_state_transition(self, line):
|
"""Apply the state_transition_rgx to a line and record state changes.
|
|
@param line: The line we're expecting to contain a state transition.
|
"""
|
state_transition_rgx = ".* ([a-zA-Z]+) updating {'([a-zA-Z]+)': ('[a-zA-Z]+'|[0-9])}.*"
|
match = apply_regex(state_transition_rgx, line)
|
if match:
|
self.line_info['state'] = match[0]
|
self.line_info['field'] = match[1]
|
self.line_info['value'] = match[2].replace("'", "")
|
|
|
def parse_line(self, line):
|
super(HostLine, self).parse_line(line)
|
|
# Regex for getting host status. Eg:
|
# 172.22.4 in Running updating {'status': 'Running'}
|
line_rgx = '.*Host (([0-9,.,a-z,-]+).*)'
|
parsed_line = apply_regex(line_rgx, self.line_info['line'])
|
if parsed_line:
|
self.line_info['line'] = parsed_line[0]
|
self.line_info['host'] = parsed_line[1]
|
self.record_state_transition(self.line_info['line'])
|
return self.format_line()
|
|
|
class HQELine(TimeLine):
|
"""Manages HQE line parsing.
|
"""
|
line_format = ('%(time)s %(hqe)s, currently in %(from_state)s, '
|
'updated to %(to_state)s. Flags: %(flags)s')
|
|
|
def record_state_transition(self, line):
|
"""Apply the state_transition_rgx to a line and record state changes.
|
|
@param line: The line we're expecting to contain a state transition.
|
"""
|
# Regex for getting hqe status. Eg:
|
# status:Running [active] -> Gathering
|
state_transition_rgx = ".*status:([a-zA-Z]+)( \[[a-z\,]+\])? -> ([a-zA-Z]+)"
|
match = apply_regex(state_transition_rgx, line)
|
if match:
|
self.line_info['from_state'] = match[0]
|
self.line_info['flags'] = match[1]
|
self.line_info['to_state'] = match[2]
|
|
|
def parse_line(self, line):
|
super(HQELine, self).parse_line(line)
|
line_rgx = r'.*\| HQE: (([0-9]+).*)'
|
parsed_line = apply_regex(line_rgx, self.line_info['line'])
|
if parsed_line:
|
self.line_info['line'] = parsed_line[0]
|
self.line_info['hqe'] = parsed_line[1]
|
self.record_state_transition(self.line_info['line'])
|
return self.format_line()
|
|
|
class LogCrawler(object):
|
"""Crawl logs.
|
|
Log crawlers are meant to apply some basic preprocessing to a log, and crawl
|
the output validating state changes. They manage line and state machine
|
creation. The initial filtering applied to the log needs to be grab all lines
|
that match an action, such as the running of a job.
|
"""
|
|
def __init__(self, log_name):
|
self.log = log_name
|
self.filter_command = 'cat %s' % log_name
|
|
|
def preprocess_log(self):
|
"""Apply some basic filtering to the log.
|
"""
|
proc = subprocess.Popen(self.filter_command,
|
shell=True, stdout=subprocess.PIPE)
|
out, err = proc.communicate()
|
return out
|
|
|
class SchedulerLogCrawler(LogCrawler):
|
"""A log crawler for the scheduler logs.
|
|
This crawler is only capable of processing information about a single job.
|
"""
|
|
def __init__(self, log_name, **kwargs):
|
super(SchedulerLogCrawler, self).__init__(log_name)
|
self.job_id = kwargs['job_id']
|
self.line_processors = [HostLine([SingleJobHostSMP()]),
|
HQELine([SingleJobHqeSMP()])]
|
self.filter_command = ('%s | grep "for job: %s"' %
|
(self.filter_command, self.job_id))
|
|
|
def parse_log(self):
|
"""Parse each line of the preprocessed log output.
|
|
Pass each line through each possible line_processor. The one that matches
|
will populate itself, call flush, this will walk the state machine of that
|
line to the next step.
|
"""
|
out = self.preprocess_log()
|
response = []
|
for job_line in out.split('\n'):
|
parsed_line = None
|
for processor in self.line_processors:
|
line = processor.parse_line(job_line)
|
if line and parsed_line:
|
raise LogLineException('Multiple Parsers claiming the line %s: '
|
'previous parsing: %s, current parsing: %s ' %
|
(job_line, parsed_line, line))
|
elif line:
|
parsed_line = line
|
try:
|
processor.flush()
|
except StateMachineViolation as e:
|
response.append(str(e))
|
raise StateMachineViolation(response)
|
response.append(parsed_line if parsed_line else job_line)
|
return response
|
|
|
def process_logs():
|
if len(sys.argv) < 2:
|
print ('Usage: ./cron_scripts/log_distiller.py 0 8415620 '
|
'You need to change the location of the log it will parse.'
|
'The job_id needs to be in the afe database.')
|
sys.exit(1)
|
|
job_id = int(sys.argv[1])
|
rpc = frontend.AFE()
|
suite_jobs = rpc.run('get_jobs', id=job_id)
|
if not suite_jobs[0]['parent_job']:
|
suite_jobs = rpc.run('get_jobs', parent_job=job_id)
|
try:
|
logfile = sys.argv[2]
|
except Exception:
|
logfile = LOGFILE
|
|
for job in suite_jobs:
|
log_crawler = SchedulerLogCrawler(logfile, job_id=job['id'])
|
for line in log_crawler.parse_log():
|
print line
|
return
|
|
|
if __name__ == '__main__':
|
process_logs()
|
