Permalink
Branch: master
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
276 lines (227 sloc) 9.51 KB
#!/usr/bin/env python
""" stresstest.py: A stress-tester for MultiprocessRotatingFileHandler
This utility spawns a bunch of processes that all try to concurrently 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
compared to see if any log messages have been lost.
"""
import os
import sys
from subprocess import call, Popen, STDOUT
from time import sleep
ROTATE_COUNT = 5000
# local lib; for testing
from mrfh import MultiprocessRotatingFileHandler
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 MultiprocessRotatingFileHandler(fn, 'a', self.rotateSize,
self.rotateCount)
# 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] %(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, un comment the next
# line. You should see this show up 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 witting 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",
description="Stress test the MRFH 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. "
"Default 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 output 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
empty = True
ai = open(a).readlines()
bi = open(b).readlines()
for line in difflib.unified_diff(ai, bi, a, b):
empty = False
out.write(line)
return empty
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 spawning 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 = [ sys.executable, 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 process 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 internal diff:")
empty = unified_diff(client_combo, shared_combo)
if empty:
print("Passed! :)")
sys.exit(0)
else:
print("Failed. :(")
sys.exit(1)
if __name__ == '__main__':
if len(sys.argv) > 1 and sys.argv[1].lower() == "client":
main_client(sys.argv[2:])
else:
main_runner(sys.argv)