import sys
import re
import exceptions
from twisted.python import log
from twisted.internet import defer
from twisted.enterprise import adbapi
from buildbot.process.buildstep import LogLineObserver
from buildbot.steps.shell import Test
class EqConnectionPool(adbapi.ConnectionPool):
"""This class works the same way as
twisted.enterprise.adbapi.ConnectionPool. But it adds the ability to
compare connection pools for equality (by comparing the arguments
passed to the constructor).
This is useful when passing the ConnectionPool to a BuildStep, as
otherwise Buildbot will consider the buildstep (and hence the
containing buildfactory) to have changed every time the configuration
is reloaded.
It also sets some defaults differently from adbapi.ConnectionPool that
are more suitable for use in MTR.
"""
def __init__(self, *args, **kwargs):
self._eqKey = (args, kwargs)
return adbapi.ConnectionPool.__init__(self,
cp_reconnect=True, cp_min=1, cp_max=3,
*args, **kwargs)
def __eq__(self, other):
if isinstance(other, EqConnectionPool):
return self._eqKey == other._eqKey
else:
return False
def __ne__(self, other):
return not self.__eq__(other)
class MtrTestFailData:
def __init__(self, testname, variant, result, info, text, callback):
self.testname = testname
self.variant = variant
self.result = result
self.info = info
self.text = text
self.callback = callback
def add(self, line):
self.text+= line
def fireCallback(self):
return self.callback(self.testname, self.variant, self.result, self.info, self.text)
class MtrLogObserver(LogLineObserver):
"""
Class implementing a log observer (can be passed to
BuildStep.addLogObserver().
It parses the output of mysql-test-run.pl as used in MySQL,
MariaDB, Drizzle, etc.
It counts number of tests run and uses it to provide more accurate
completion estimates.
It parses out test failures from the output and summarises the results on
the Waterfall page. It also passes the information to methods that can be
overridden in a subclass to do further processing on the information."""
_line_re = re.compile(r"^([-._0-9a-zA-z]+)( '[-_ a-zA-Z]+')?\s+(w[0-9]+\s+)?\[ (fail|pass) \]\s*(.*)$")
_line_re2 = re.compile(r"^[-._0-9a-zA-z]+( '[-_ a-zA-Z]+')?\s+(w[0-9]+\s+)?\[ [-a-z]+ \]")
_line_re3 = re.compile(r"^\*\*\*Warnings generated in error logs during shutdown after running tests: (.*)")
_line_re4 = re.compile(r"^The servers were restarted [0-9]+ times$")
_line_re5 = re.compile(r"^Only\s+[0-9]+\s+of\s+[0-9]+\s+completed.$")
def __init__(self, textLimit=5, testNameLimit=16, testType=None):
self.textLimit = textLimit
self.testNameLimit = testNameLimit
self.testType = testType
self.numTests = 0
self.testFail = None
self.failList = []
self.warnList = []
LogLineObserver.__init__(self)
def setLog(self, loog):
LogLineObserver.setLog(self, loog)
d= loog.waitUntilFinished()
d.addCallback(lambda l: self.closeTestFail())
def outLineReceived(self, line):
stripLine = line.strip("\r\n")
m = self._line_re.search(stripLine)
if m:
testname, variant, worker, result, info = m.groups()
self.closeTestFail()
self.numTests += 1
self.step.setProgress('tests', self.numTests)
if result == "fail":
if variant == None:
variant = ""
else:
variant = variant[2:-1]
self.openTestFail(testname, variant, result, info, stripLine + "\n")
else:
m = self._line_re3.search(stripLine)
if m:
stuff = m.group(1)
self.closeTestFail()
testList = stuff.split(" ")
self.doCollectWarningTests(testList)
elif (self._line_re2.search(stripLine) or
self._line_re4.search(stripLine) or
self._line_re5.search(stripLine) or
stripLine == "Test suite timeout! Terminating..." or
stripLine.startswith("mysql-test-run: *** ERROR: Not all tests completed") or
(stripLine.startswith("------------------------------------------------------------")
and self.testFail != None)):
self.closeTestFail()
else:
self.addTestFailOutput(stripLine + "\n")
def openTestFail(self, testname, variant, result, info, line):
self.testFail = MtrTestFailData(testname, variant, result, info, line, self.doCollectTestFail)
def addTestFailOutput(self, line):
if self.testFail != None:
self.testFail.add(line)
def closeTestFail(self):
if self.testFail != None:
self.testFail.fireCallback()
self.testFail = None
def addToText(self, src, dst):
lastOne = None
count = 0
for t in src:
if t != lastOne:
dst.append(t)
count += 1
if count >= self.textLimit:
break
def makeText(self, done):
if done:
text = ["test"]
else:
text = ["testing"]
if self.testType:
text.append(self.testType)
fails = self.failList[:]
fails.sort()
self.addToText(fails, text)
warns = self.warnList[:]
warns.sort()
self.addToText(warns, text)
return text
# Update waterfall status.
def updateText(self):
self.step.step_status.setText(self.makeText(False))
strip_re = re.compile(r"^[a-z]+\.")
def displayTestName(self, testname):
displayTestName = self.strip_re.sub("", testname)
if len(displayTestName) > self.testNameLimit:
displayTestName = displayTestName[:(self.testNameLimit-2)] + "..."
return displayTestName
def doCollectTestFail(self, testname, variant, result, info, text):
self.failList.append("F:" + self.displayTestName(testname))
self.updateText()
self.collectTestFail(testname, variant, result, info, text)
def doCollectWarningTests(self, testList):
for t in testList:
self.warnList.append("W:" + self.displayTestName(t))
self.updateText()
self.collectWarningTests(testList)
# These two methods are overridden to actually do something with the data.
def collectTestFail(self, testname, variant, result, info, text):
pass
def collectWarningTests(self, testList):
pass
class MTR(Test):
"""
Build step that runs mysql-test-run.pl, as used in MySQL, Drizzle,
MariaDB, etc.
It uses class MtrLogObserver to parse test results out from the
output of mysql-test-run.pl, providing better completion time
estimates and summarising test failures on the waterfall page.
It also provides access to mysqld server error logs from the test
run to help debugging any problems.
Optionally, it can insert into a database data about the test run,
including details of any test failures.
Parameters:
textLimit
Maximum number of test failures to show on the waterfall page
(to not flood the page in case of a large number of test
failures. Defaults to 5.
testNameLimit
Maximum length of test names to show unabbreviated in the
waterfall page, to avoid excessive column width. Defaults to 16.
parallel
Value of --parallel option used for mysql-test-run.pl (number
of processes used to run the test suite in parallel). Defaults
to 4. This is used to determine the number of server error log
files to download from the slave. Specifying a too high value
does not hurt (as nonexisting error logs will be ignored),
however if using --parallel value greater than the default it
needs to be specified, or some server error logs will be
missing.
dbpool
An instance of twisted.enterprise.adbapi.ConnectionPool, or None.
Defaults to None. If specified, results are inserted into the database
using the ConnectionPool.
The class process.mtrlogobserver.EqConnectionPool subclass of
ConnectionPool can be useful to pass as value for dbpool, to
avoid having config reloads think the Buildstep is changed
just because it gets a new ConnectionPool instance (even
though connection parameters are unchanged).
autoCreateTables
Boolean, defaults to False. If True (and dbpool is specified), the
necessary database tables will be created automatically if they do
not exist already. Alternatively, the tables can be created manually
from the SQL statements found in the mtrlogobserver.py source file.
test_type
test_info
Two descriptive strings that will be inserted in the database tables if
dbpool is specified. The test_type string, if specified, will also
appear on the waterfall page."""
def __init__(self, dbpool=None, test_type=None, test_info="",
description=None, descriptionDone=None,
autoCreateTables=False, textLimit=5, testNameLimit=16,
parallel=4, logfiles = {}, lazylogfiles = True,
warningPattern="MTR's internal check of the test case '.*' failed",
mtr_subdir="mysql-test", **kwargs):
if description is None:
description = ["testing"]
if test_type:
description.append(test_type)
if descriptionDone is None:
descriptionDone = ["test"]
if test_type:
descriptionDone.append(test_type)
Test.__init__(self, logfiles=logfiles, lazylogfiles=lazylogfiles,
description=description, descriptionDone=descriptionDone,
warningPattern=warningPattern, **kwargs)
self.dbpool = dbpool
self.test_type = test_type
self.test_info = test_info
self.autoCreateTables = autoCreateTables
self.textLimit = textLimit
self.testNameLimit = testNameLimit
self.parallel = parallel
self.mtr_subdir = mtr_subdir
self.progressMetrics += ('tests',)
self.addFactoryArguments(dbpool=self.dbpool,
test_type=self.test_type,
test_info=self.test_info,
autoCreateTables=self.autoCreateTables,
textLimit=self.textLimit,
testNameLimit=self.testNameLimit,
parallel=self.parallel,
mtr_subdir=self.mtr_subdir)
def start(self):
properties = self.build.getProperties()
subdir = properties.render(self.mtr_subdir)
# Add mysql server logfiles.
for mtr in range(0, self.parallel+1):
for mysqld in range(1, 4+1):
if mtr == 0:
logname = "mysqld.%d.err" % mysqld
filename = "var/log/mysqld.%d.err" % mysqld
else:
logname = "mysqld.%d.err.%d" % (mysqld, mtr)
filename = "var/%d/log/mysqld.%d.err" % (mtr, mysqld)
self.addLogFile(logname, subdir + "/" + filename)
self.myMtr = self.MyMtrLogObserver(textLimit=self.textLimit,
testNameLimit=self.testNameLimit,
testType=self.test_type)
self.addLogObserver("stdio", self.myMtr)
# Insert a row for this test run into the database and set up
# build properties, then start the command proper.
d = self.registerInDB()
d.addCallback(self.afterRegisterInDB)
d.addErrback(self.failed)
def getText(self, command, results):
return self.myMtr.makeText(True)
def runInteractionWithRetry(self, actionFn, *args, **kw):
"""
Run a database transaction with dbpool.runInteraction, but retry the
transaction in case of a temporary error (like connection lost).
This is needed to be robust against things like database connection
idle timeouts.
The passed callable that implements the transaction must be retryable,
ie. it must not have any destructive side effects in the case where
an exception is thrown and/or rollback occurs that would prevent it
from functioning correctly when called again."""
def runWithRetry(txn, *args, **kw):
retryCount = 0
while(True):
try:
return actionFn(txn, *args, **kw)
except txn.OperationalError:
retryCount += 1
if retryCount >= 5:
raise
excType, excValue, excTraceback = sys.exc_info()
log.msg("Database transaction failed (caught exception %s(%s)), retrying ..." % (excType, excValue))
txn.close()
txn.reconnect()
txn.reopen()
return self.dbpool.runInteraction(runWithRetry, *args, **kw)
def runQueryWithRetry(self, *args, **kw):
"""
Run a database query, like with dbpool.runQuery, but retry the query in
case of a temporary error (like connection lost).
This is needed to be robust against things like database connection
idle timeouts."""
def runQuery(txn, *args, **kw):
txn.execute(*args, **kw)
return txn.fetchall()
return self.runInteractionWithRetry(runQuery, *args, **kw)
def registerInDB(self):
if self.dbpool:
return self.runInteractionWithRetry(self.doRegisterInDB)
else:
return defer.succeed(0)
# The real database work is done in a thread in a synchronous way.
def doRegisterInDB(self, txn):
# Auto create tables.
# This is off by default, as it gives warnings in log file
# about tables already existing (and I did not find the issue
# important enough to find a better fix).
if self.autoCreateTables:
txn.execute("""
CREATE TABLE IF NOT EXISTS test_run(
id INT PRIMARY KEY AUTO_INCREMENT,
branch VARCHAR(100),
revision VARCHAR(32) NOT NULL,
platform VARCHAR(100) NOT NULL,
dt TIMESTAMP NOT NULL,
bbnum INT NOT NULL,
typ VARCHAR(32) NOT NULL,
info VARCHAR(255),
KEY (branch, revision),
KEY (dt),
KEY (platform, bbnum)
) ENGINE=innodb
""")
txn.execute("""
CREATE TABLE IF NOT EXISTS test_failure(
test_run_id INT NOT NULL,
test_name VARCHAR(100) NOT NULL,
test_variant VARCHAR(16) NOT NULL,
info_text VARCHAR(255),
failure_text TEXT,
PRIMARY KEY (test_run_id, test_name, test_variant)
) ENGINE=innodb
""")
txn.execute("""
CREATE TABLE IF NOT EXISTS test_warnings(
test_run_id INT NOT NULL,
list_id INT NOT NULL,
list_idx INT NOT NULL,
test_name VARCHAR(100) NOT NULL,
PRIMARY KEY (test_run_id, list_id, list_idx)
) ENGINE=innodb
""")
revision = None
try:
revision = self.getProperty("got_revision")
except exceptions.KeyError:
revision = self.getProperty("revision")
typ = "mtr"
if self.test_type:
typ = self.test_type
txn.execute("""
INSERT INTO test_run(branch, revision, platform, dt, bbnum, typ, info)
VALUES (%s, %s, %s, CURRENT_TIMESTAMP(), %s, %s, %s)
""", (self.getProperty("branch"), revision,
self.getProperty("buildername"), self.getProperty("buildnumber"),
typ, self.test_info))
return txn.lastrowid
def afterRegisterInDB(self, insert_id):
self.setProperty("mtr_id", insert_id)
self.setProperty("mtr_warn_id", 0)
Test.start(self)
def reportError(self, err):
log.msg("Error in async insert into database: %s" % err)
class MyMtrLogObserver(MtrLogObserver):
def collectTestFail(self, testname, variant, result, info, text):
# Insert asynchronously into database.
dbpool = self.step.dbpool
run_id = self.step.getProperty("mtr_id")
if dbpool == None:
return defer.succeed(None)
if variant == None:
variant = ""
d = self.step.runQueryWithRetry("""
INSERT INTO test_failure(test_run_id, test_name, test_variant, info_text, failure_text)
VALUES (%s, %s, %s, %s, %s)
""", (run_id, testname, variant, info, text))
d.addErrback(self.step.reportError)
return d
def collectWarningTests(self, testList):
# Insert asynchronously into database.
dbpool = self.step.dbpool
if dbpool == None:
return defer.succeed(None)
run_id = self.step.getProperty("mtr_id")
warn_id = self.step.getProperty("mtr_warn_id")
self.step.setProperty("mtr_warn_id", warn_id + 1)
q = ("INSERT INTO test_warnings(test_run_id, list_id, list_idx, test_name) " +
"VALUES " + ", ".join(map(lambda x: "(%s, %s, %s, %s)", testList)))
v = []
idx = 0
for t in testList:
v.extend([run_id, warn_id, idx, t])
idx = idx + 1
d = self.step.runQueryWithRetry(q, tuple(v))
d.addErrback(self.step.reportError)
return d
|