Merge lp:~vila/udd/analyze_log into lp:udd

[Vincent Ladeuil]

I did some log analysis with this script and thought it can help us in the future.

It will probably have a short live for some parts of it once we can provide more data from the imports themselves but in the mean time there is no real value to keep it private.

[Martin Pool (mbp)]

Please mention it in a HACKING file so that people can find it in future.

review: Approve

[lp:~vila/udd/analyze_log]

398. By Vincent Ladeuil on 2011-02-18

Add analyze_log to summarize past activities

[Vincent Ladeuil (vila)]

>>>>> Martin Pool <email address hidden> writes:

    > Review: Approve
    > Please mention it in a HACKING file so that people can find it in future.

Done: https://wiki.ubuntu.com/DistributedDevelopment/UnderTheHood/Importer/Operational

=== added file 'analyze_log.py'
--- analyze_log.py	1970-01-01 00:00:00 +0000
+++ analyze_log.py	2011-02-15 09:17:57 +0000
@@ -0,0 +1,360 @@
+#!/usr/bin/python
+"""Analyzes the package importer log files collecting various historical data.
+
+This may need to be tweaked as the logger usage evolves.
+"""
+
+import datetime
+import errno
+import operator
+import os
+import re
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(__file__))
+import icommon
+
+
+class LoggerLineMatcher(object):
+
+    actor = None
+
+    def __init__(self, actor, ident, regexp, args=None):
+        """Logger line scanner helper.
+
+        :param actor: Uniquely identify an actor in the log file.
+
+        :param ident: Uniquely identify a logger line type for agiven ``actor``.
+
+        :param regexp: Regular expression the message should match.
+
+        :param args: A list a indices in the groups() matched, each
+            corresponding group will be added in the order given in the
+            details().
+        """
+        self.actor = actor
+        self.ident = ident
+        self.regexp = re.compile(regexp)
+        self.args = args
+
+    def matches(self, msg):
+        self.match = self.regexp.match(msg)
+        return self.match is not None
+
+
+class ActorParser(object):
+    """A parser for an actor defined by the lines it emits in a log file."""
+
+    def __init__(self, ident):
+        self.ident = ident
+        self.start = None # datetime.datetime object
+        self.end = None # datetime.datetime object
+        self.duration = None #datetime.timedelta object
+
+    def see(self, state, stamp, args):
+        raise NotImplementedError(self.see)
+
+    def dead(self):
+        return self.end is not None
+
+    def round_time(self, t):
+        return t - datetime.timedelta(microseconds=t.microsecond)
+
+    def round_delta(self, d):
+        return d - datetime.timedelta(microseconds=d.microseconds)
+
+    def report_key(self):
+        return (self.end, self.start)
+
+    def report_times(self, out, epoch=None, apocalypse=None):
+        estimate = False
+        if self.start is None:
+            estimate = True
+            self.start = epoch
+        out.write('%s%s' % ([' ', '<'][estimate],
+                                  self.round_time(self.start)))
+        if self.end is None:
+            estimate = True
+            self.end = apocalypse
+        duration = self.end - self.start
+        if duration.seconds >= 10:
+            # Lost in the noise
+            duration = self.round_delta(duration)
+        self.duration = duration
+        out.write(' %s%s' % ([' ', '>'][estimate], duration))
+        out.write(' %s\n' % (self.ident,))
+
+
+class ControllerParser(ActorParser):
+
+    def __init__(self):
+        super(ControllerParser, self).__init__('controller')
+        self.nb_packages = 0
+        self.cumulated_duration = datetime.timedelta(0)
+        self.longest = None
+        self.longest_duration = datetime.timedelta(0)
+
+    def see(self, state, stamp, args):
+        if state == 'starting':
+            self.start = stamp
+        elif state == 'signaled':
+            pass
+        elif state == 'waiting':
+            pass
+        else:
+            assert state == 'finished'
+            self.end = stamp
+
+    def package_ends(self, pack):
+        if pack.duration > self.longest_duration:
+            self.longest = pack.ident
+            self.longest_duration = pack.duration
+        self.cumulated_duration += pack.duration
+        self.nb_packages += 1
+
+    def report_times(self, out, epoch=None, apocalypse=None):
+        super(ControllerParser, self).report_times(out, epoch, apocalypse)
+        if self.nb_packages == 0:
+            self.nb_packages = 1 # avoid divide by zero
+        out.write('  imported %6d, average %10s, longest: %10s %s\n'
+                  % (self.nb_packages, self.round_delta(self.cumulated_duration
+                                                        / self.nb_packages),
+                     self.round_delta(self.longest_duration), self.longest,
+                     ))
+
+
+class DriverParser(ActorParser):
+
+    def __init__(self):
+        super(DriverParser, self).__init__('driver')
+        self.concurrency = 0
+
+    def see(self, state, stamp, args):
+        if state == 'starting':
+            self.start = stamp
+        elif state == 'active':
+            concurrency = len(args[0].split(','))
+            if concurrency != self.concurrency:
+                self.concurrency = concurrency
+                print '>>> %s - Concurrency is now: %d' % (stamp, concurrency,)
+        elif state in ('asked_to_stop', 'crashed'):
+            pass
+        else:
+            assert state in ('stopping', 'finished')
+            self.end = stamp
+
+
+class PackageParser(ActorParser):
+
+    def __init__(self, name):
+        super(PackageParser, self).__init__(name)
+        self.tb = None
+
+    def see(self, state, stamp, args):
+        if state == 'trying':
+            self.start = stamp
+        elif state == 'locked':
+            self.end = stamp
+        elif state == 'finished':
+            pass
+        elif state == 'success':
+            # There is something to grab in args there, distro, pocket, etc
+            self.end = stamp
+        else:
+            assert state == 'failed'
+            self.end = stamp
+            # This is followed by "something" containing a
+            # traceback, so we slurp all lines until the next valid
+            # logger line
+            log = args[0]
+            something = []
+            while True:
+                lineno, line = log.next_line()
+                # Trying to verify that ``line`` is a proper logger
+                # line is far too expensive, so we try a cheap
+                # substitute instead the initial date will start
+                # with 20 for the current century.
+                if line.startswith('20'):
+                    log.pushed_back_line = line
+                    break
+                something.append(line)
+            self.tb = ''.join(something)
+
+
+class Log(object):
+
+    def __init__(self, path):
+        self.path = path
+        self.stream = iter(open(self.path))
+        self.lineno = 0
+        self.pushed_back_line = None
+        # The first and last stamp seen in the log file
+        self.epoch = None
+        self.apocalypse = None
+        # A logger line is:
+        #  <time stamp> - <component?> - level - msg
+        self.logger_line_re = re.compile(
+            '([0-9- :,]{23}) - ([^ -]+) - ([A-Z]+) - (.*)')
+
+        def Controller(ident, regexp):
+            return LoggerLineMatcher('controller', ident, regexp)
+
+        def Driver(ident, regexp):
+            return LoggerLineMatcher('driver', ident, regexp)
+
+        def Package(ident, regexp):
+            return LoggerLineMatcher('package', ident, regexp)
+
+
+        # All the possible lines, most often found first (as a cheap shot
+        # optimization).
+        self.matchers = [
+            Package('trying', 'Trying ([^ ]+)'),
+            Package('locked', "Couldn't lock ([^ ]+), skipping"),
+            Package('success', 'Success ([^ ]+): (.*)'),
+            Package('finished', 'thread for ([^ ]+) finished'),
+            Package('failed', 'Importing ([^ ]+) failed:'),
+            # Driver logger lines
+            Driver('active', 'threads for \[([^\]]+)\] still active'),
+            Driver('asked_to_stop', 'Driver asked to stop'),
+            Driver('crashed', 'Driver hit (.*)'),
+            Driver('stopping', 'Driver stopping'),
+            Driver('finished', 'Driver finished: stopping'),
+            # Controller logger lines
+            Controller('starting', 'Starting up'),
+            Controller('signaled', 'Received signal'),
+            Controller('waiting', 'Waiting for driver to finish'),
+            Controller('finished', 'Finished'),
+            ]
+        # At any given point in time, there could be one controller, one driver
+        # and several packages writing to the log file. There is no guarantee
+        # that any of them will have corresponding lines for starting or
+        # finishing (not to mention dieing).
+        self.controller = None
+        self.driver = None
+        self.package_parsers = {}
+
+    def scan_logger_line(self, stamp_str, component, level, msg):
+        stamp = datetime.datetime.strptime(stamp_str[:-4], '%Y-%m-%d %H:%M:%S')
+        # strptime can't parse the miliseconds, do it manually
+        stamp = stamp.replace(microsecond=int(stamp_str[-3:])*1000)
+        if self.epoch is None:
+            self.epoch = stamp
+        self.apocalypse = stamp
+
+        for m in self.matchers:
+            if m.matches(msg):
+                return (stamp, m.actor, m.ident, m.match.groups())
+
+        return (stamp, 'unknown', 'garbage', rest)
+
+    def next_line(self):
+        if self.pushed_back_line is not None:
+            # Give back the pushed back line without changing the lineno
+            line = self.pushed_back_line
+            self.pushed_back_line = None
+        else:
+            line = self.stream.next()
+            self.lineno +=1
+        return self.lineno, line
+
+    def scan(self):
+        while True:
+            lineno, line = self.next_line()
+            res = self.logger_line_re.match(line)
+            if res is not None:
+                stamp, component, level, msg = res.groups()
+                yield lineno, 'logger', self.scan_logger_line(
+                    stamp, component, level, msg)
+            else:
+                yield lineno, 'garbage', line
+
+    def get_controller(self):
+        # autovivification
+        if self.controller is None:
+            self.controller = ControllerParser()
+        return self.controller
+
+    def get_driver(self):
+        # autovivification
+        if self.driver is None:
+            self.driver = DriverParser()
+        return self.driver
+
+    def get_package_parser(self, name):
+        try:
+            p = self.package_parsers[name]
+        except KeyError:
+            # autovivification
+            p = PackageParser(name)
+            self.package_parsers[name] = p
+        return p
+
+    def report_all(self):
+        c = self.get_controller()
+        packages = sorted(self.package_parsers.values(),
+                          key=operator.methodcaller('report_key'))
+        for p in packages:
+            # Sort by p.start (including None)
+            p.report_times(sys.stdout, self.epoch, self.apocalypse)
+            c.package_ends(p)
+        self.get_driver().report_times(sys.stdout, self.epoch, self.apocalypse)
+        c.report_times(sys.stdout, self.epoch, self.apocalypse)
+
+    def parse(self):
+        # controller and driver always run together (there is a bug if the
+        # controller is killed with a hammer, we don't care.
+        self.get_controller()
+        self.get_driver()
+        try:
+            for lineno, typ, args in self.scan():
+                if typ == 'logger':
+                    (stamp, actor, ident,
+                     remaining) = args[0], args[1], args[2], args[3]
+                    if actor == 'controller':
+                        c = self.get_controller()
+                        c.see(ident, stamp, remaining)
+                        if c.dead():
+                            self.report_all()
+                            self.package_parsers = {}
+                            self.controller = self.driver = None
+                    elif actor == 'driver':
+                        d = self.get_driver()
+                        d.see(ident, stamp, remaining)
+                    elif actor == 'package':
+                        name = remaining[0]
+                        p = self.get_package_parser(name)
+                        p.see(ident, stamp, [self] + list(remaining[1:]))
+                        if p.dead():
+                            p.report_times(sys.stdout, self.epoch)
+                            self.get_controller().package_ends(p)
+                            del self.package_parsers[name]
+                else:
+                    line = args
+                    if line != '\n': # Don't bother reporting empty lines
+                        # Unrecognized line type
+                        sys.stdout.write('%s:%s garbage: %s'
+                                         % (self.path, lineno, args,))
+        finally:
+            self.stream.close()
+        # Final reports can break with log files without a single logger line
+        # (no time stamps seen, we don't care)
+        self.report_all()
+
+def main(args):
+    if len(args) != 2:
+        print 'Usage: %s <log file path>'
+        exit(1)
+
+    log_path = args[1]
+    if not os.path.exists(log_path):
+        print '%s does not exist' % (log_path,)
+        exit(1)
+
+    log = Log(log_path)
+    log.parse()
+
+
+if __name__ == '__main__':
+    main(sys.argv)