stresstest.py

#!/usr/bin/env python
""" stresstest.py:  A stres-tester for ConcurrentRotatingFileHandler

This utility spawns a bunch of processes that all try to concurently write to
the same file. This is pretty much the worst-case scenario for my log handler.
Once all of the processes have completed writing to the log file, the output is
complared to see if any log messages have been lost.

In the future, I may also add in support for testing with each process having
multiple threads.


"""

__version__ = '$Id: stresstest.py 5481 2008-05-21 15:18:22Z lowell $'
__author__ = 'Lowell Alleman'


import os
import sys
from subprocess import call, Popen, STDOUT
from time import sleep

ROTATE_COUNT = 5000

# local lib; for testing
from cloghandler import ConcurrentRotatingFileHandler

class RotateLogStressTester:
    def __init__(self, sharedfile, uniquefile, name="LogStressTester"):
        self.sharedfile = sharedfile
        self.uniquefile = uniquefile
        self.name = name
        self.writeLoops = 100000
        self.rotateSize = 128 * 1024
        self.rotateCount = ROTATE_COUNT
        self.random_sleep_mode = False
        self.debug = False
    
    def getLogHandler(self, fn):
        """ Override this method if you want to test a different logging handler
        class. """
        return ConcurrentRotatingFileHandler(fn, 'a', self.rotateSize,
                                             self.rotateCount, debug=self.debug)
        # To run the test with the standard library's RotatingFileHandler:
        # from logging.handlers import RotatingFileHandler
        # return RotatingFileHandler(fn, 'a', self.rotateSize, self.rotateCount)
    
    def start(self):
        from logging import getLogger, FileHandler, Formatter, DEBUG
        self.log = getLogger(self.name)
        self.log.setLevel(DEBUG)
        
        formatter = Formatter('%(asctime)s [%(process)d:%(threadName)s] %(levelname)-8s %(name)s:  %(message)s')
        # Unique log handler (single file)
        handler  = FileHandler(self.uniquefile, "w")
        handler.setLevel(DEBUG)
        handler.setFormatter(formatter)
        self.log.addHandler(handler)
        
        # If you suspect that the diff stuff isn't working, uncomment the next
        # line.  You should see this showup once per-process.
        # self.log.info("Here is a line that should only be in the first output.")
        
        # Setup output used for testing
        handler = self.getLogHandler(self.sharedfile)
        handler.setLevel(DEBUG)
        handler.setFormatter(formatter)
        self.log.addHandler(handler)
        
        # If this ever becomes a real "Thread", then remove this line:
        self.run()
    
    def run(self):
        c = 0
        from random import choice, randint
        # Use a bunch of random quotes, numbers, and severity levels to mix it up a bit!
        msgs = ["I found %d puppies", "There are %d cats in your hatz",
                "my favorite number is %d", "I am %d years old.", "1 + 1 = %d",
                "%d/0 = DivideByZero", "blah!  %d thingies!", "8 15 16 23 48 %d",
                "the worlds largest prime number: %d", "%d happy meals!"]
        logfuncts = [self.log.debug, self.log.info, self.log.warn, self.log.error]
        
        self.log.info("Starting to write random log message.   Loop=%d", self.writeLoops)
        while c <= self.writeLoops:
            c += 1
            msg = choice(msgs)
            logfunc = choice(logfuncts)
            logfunc(msg, randint(0,99999999))
            
            if self.random_sleep_mode and c % 1000 == 0:
                # Sleep from 0-15 seconds
                s = randint(1,15)
                print "PID %d sleeping for %d seconds" % (os.getpid(), s)
                sleep(s)
            # break
        self.log.info("Done writting random log messages.")

def iter_lognames(logfile, count):
    """ Generator for log file names based on a rotation scheme """
    for i in range(count -1, 0, -1):
        yield "%s.%d" % (logfile, i)
    yield logfile

def iter_logs(iterable, missing_ok=False):
    """ Generator to extract log entries from shared log file. """
    for fn in iterable:
        if os.path.exists(fn):
            for line in open(fn):
                yield line
        elif not missing_ok:
            raise ValueError("Missing log file %s" % fn)

def combine_logs(combinedlog, iterable, mode="w"):
    """ write all lines (iterable) into a single log file. """
    fp = open(combinedlog, mode)
    for chunk in iterable:
        fp.write(chunk)
    fp.close()



from optparse import OptionParser
parser = OptionParser(usage="usage:  %prog",
                      version=__version__,
                      description="Stress test the safelogrotate module.")
parser.add_option("--log-calls", metavar="NUM",
                  action="store", type="int", default=50000,
                  help="Number of logging entries to write to each log file.  "
                  "Deafult is %default")
parser.add_option("--random-sleep-mode",
                  action="store_true", default=False)
parser.add_option("--debug",
                  action="store_true", default=False)


def main_client(args):
    (options, args) = parser.parse_args(args)
    if len(args) != 2:
        raise ValueError("Require 2 arguments.  We have %d args" % len(args))
    (shared, client) = args
    
    if os.path.isfile(client):
        sys.stderr.write("Already a client using ouptut file %s\n" % client)
        sys.exit(1)
    tester = RotateLogStressTester(shared, client)
    tester.random_sleep_mode = options.random_sleep_mode
    tester.debug = options.debug
    tester.writeLoops = options.log_calls
    tester.start()
    print "We are done  pid=%d" % os.getpid()



class TestManager:
    class ChildProc(object):
        """ Very simple child container class."""
        __slots__ = [ "popen", "sharedfile", "clientfile" ]
        def __init__(self, **kwargs):
            self.update(**kwargs)
        def update(self, **kwargs):
            for key, val in kwargs.items():
                setattr(self, key, val)
    
    def __init__(self):
        self.tests = []
    
    def launchPopen(self, *args, **kwargs):
        proc = Popen(*args, **kwargs)
        cp = self.ChildProc(popen=proc)
        self.tests.append(cp)
        return cp
    
    def wait(self, check_interval=3):
        """ Wait for all child test processes to complete. """
        print "Waiting while children are out running and playing!"
        while True:
            sleep(check_interval)
            waiting = []
            for cp in self.tests:
                if cp.popen.poll() is None:
                    waiting.append(cp.popen.pid)
            if not waiting:
                break
            print "Waiting on %r " % waiting
        print "All children have stopped."
    
    def checkExitCodes(self):
        for cp in self.tests:
            if cp.popen.poll() != 0:
                return False
        return True



def unified_diff(a,b, out=sys.stdout):
    import difflib
    ai = open(a).readlines()
    bi = open(b).readlines()
    for line in difflib.unified_diff(ai, bi, a, b):
        out.write(line)



def main_runner(args):
    parser.add_option("--processes", metavar="NUM",
                      action="store", type="int", default=3,
                      help="Number of processes to spawn.  Default: %default")
    parser.add_option("--delay", metavar="secs",
                      action="store", type="float", default=2.5,
                      help="Wait SECS before spawing next processes.  "
                      "Default: %default")
    parser.add_option("-p", "--path", metavar="DIR",
                      action="store", default="test",
                      help="Path to a temporary directory.  Default: '%default'")
    
    
    this_script = args[0]
    (options, args) = parser.parse_args(args)
    options.path = os.path.abspath(options.path)
    if not os.path.isdir(options.path):
        os.makedirs(options.path)
    
    manager = TestManager()
    shared = os.path.join(options.path, "shared.log")
    for client_id in range(options.processes):
        client = os.path.join(options.path, "client.log_client%s.log" % client_id)
        cmdline = ["python", this_script, "client", shared, client,
                   "--log-calls=%d" % options.log_calls ]
        if options.random_sleep_mode:
            cmdline.append("--random-sleep-mode")
        if options.debug:
            cmdline.append("--debug")
        
        child = manager.launchPopen(cmdline)
        child.update(sharedfile=shared, clientfile=client)
        sleep(options.delay)
    
    # Wait for all of the subprocesses to exit
    manager.wait()
    # Check children exit codes
    if not manager.checkExitCodes():
        sys.stderr.write("One or more of the child proceses has failed.\n"
                         "Aborting test.\n")
        sys.exit(2)
    
    client_combo = os.path.join(options.path, "client.log.combo")
    shared_combo = os.path.join(options.path, "shared.log.combo")
    
    # Combine all of the log files...
    client_files = [ child.clientfile for child in manager.tests ]
        
    if False:
        def sort_em(iterable):
            return iterable
    else:
       sort_em = sorted
    
    print "Writing out combined client logs..." ,
    combine_logs(client_combo, sort_em(iter_logs(client_files)))
    print "done."
    
    print "Writing out combined shared logs..." ,
    shared_log_files = iter_lognames(shared, ROTATE_COUNT)
    log_lines = iter_logs(shared_log_files, missing_ok=True)
    combine_logs(shared_combo, sort_em(log_lines))
    print "done."
    
    print "Running interal diff:  (If the next line is 'end of diff', then the stress test passed!)"
    unified_diff(client_combo, shared_combo)
    print "   --- end of diff ----"



if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1].lower() == "client":
            main_client(sys.argv[2:])
    else:
        main_runner(sys.argv)