A basic MySQL replication monitor using python-bottle

python-mysql-repl-monitor

#!/usr/bin/python
"""
python-bottle based MySQL replication monitor.
Configuration is specified via [server:<server_name>] and [bottle] directives
in /etc/rack_replmon/monitor.conf. The monitor will parse any entry that
begins with ^server as directives for a server to scan

Valid [server:<server_name>] directives:
host, port, user, passwd

Valid [bottle] directives:
host, port

If any server fails to meet the following replication requirements,
an error is returned:

- Slave lag > 60s
- Slave_IO_Thread not running
- Slave_SQL_Thread not running

If all tests succeed, the first line will match ^OK:
If any tests fail, the first line will match ^BAD:
"""

from bottle import route, run, template
import MySQLdb
import ConfigParser
import sys


class Config(ConfigParser.ConfigParser):

    def __getitems__(self, section):
        try:
            return dict(self.items(section))
        except ConfigParser.NoSectionError:
            raise KeyError(section)
            
            
def test_slave(server):
    global config

    dbh = None
    dsn = {"host" : "localhost", "port" : 3306,
           "user" : "root", "passwd" : ""}
    try:
        dsn.update(config.__getitems__(server))
    except KeyError, exc:
        return (1, "Server %s not found in configuration" % server)

    try:
        dbh = MySQLdb.connect(host=dsn["host"], port=int(dsn["port"]),
                              user=dsn["user"], passwd=dsn["passwd"]);
        cur = dbh.cursor(MySQLdb.cursors.DictCursor)

        cur.execute("SHOW SLAVE STATUS")
        slave_status = cur.fetchone()

        if slave_status["Seconds_Behind_Master"] <= 60 and slave_status["Slave_IO_Running"] == "Yes" and slave_status["Slave_SQL_Running"] == "Yes":
            return (0, "+ Checks for server %s:%s completed successfully" % (dsn["host"], dsn["port"]))

        else:
            fail_print = " Relay coordinates: %s %d\n Master coordinates: %s %d\n IO Error: %d -- %s\n SQL Error: %d -- %s" % (slave_status["Relay_Log_File"], slave_status["Relay_Log_Pos"], slave_status["Relay_Master_Log_File"], slave_status["Exec_Master_Log_Pos"], slave_status["Last_IO_Errno"], slave_status["Last_IO_Error"], slave_status["Last_SQL_Errno"], slave_status["Last_SQL_Error"])
            return (1, "x Replication Error: Checks failed for server %s:%s\n%s" % (dsn["host"], dsn["port"], fail_print))

    except MySQLdb.MySQLError, exc:
        return (1, "x Connection Error %d: %s" % (exc.args[0],exc.args[1]))

    finally:
        if dbh:
            dbh.close()


def enumerate_slaves(slave_filter):
    global config
    test_success = 1
    test_status_list = []

    if slave_filter == None:
        server_list = [server for server in config.sections() if server.startswith("server")]
    else:
        slave_filter = "server:" + slave_filter
        if slave_filter in config.sections():
            server_list = [slave_filter]
        else:
            return "Server %s not found in configuration\n" % slave_filter

    for server in server_list:
        (test_return, status_string) = test_slave(server)
        if test_return != 0:
            test_success = 0
        test_status_list.append(status_string)
    
    if test_success == 1:
        test_status_list.insert(0, "OK: All checks completed successfully")
    else:
        test_status_list.insert(0, "BAD: At least one check failed")

    return '\n'.join(test_status_list) + '\n'
    

def main():
    global config
    config = Config()

    try:
        config.readfp(open("/etc/rack_replmon/monitor.conf"))
    except ConfigParser.Error, exc:
        print >>sys.stderr, "Failed to parse config ", exc
        return 1
    except IOError, exc:
        print >>sys.stderr, "Failed to open config ", exc
        return 1

    @route('/<server>')
    @route('/')
    def index(server=None):
        if server == None:
            slave_status = enumerate_slaves(None)
        else:
            slave_status = enumerate_slaves(server)
        return template('{{slave_status}}', slave_status=slave_status)

    bottle_server = {"host" : "localhost", "port" : 8080}
    if "bottle" in config.sections():
        bottle_server.update(config.__getitems__("bottle"))
    run(host=bottle_server["host"], port=bottle_server["port"])

    return 0

if __name__ == '__main__':
    sys.exit(main())

From here, I have the following goals outlined next:

• Get tests built (I was looking at using nosetest. Is there anything else I should consider using?)

I am a big fan of nosetests for unit testing. python also has a unittest module in the stdlib, but it requires a lot more boiler plate and nosetests can also run those tests through its interface as well. I only use unittest if i'm integrating some existing test case (sometimes from a python backport for compatibility, etc.), but always test with the nosetest frontend.

Some other tools that are very useful to look at that I use frequently:

• pyflakes - this is effectively a pep8 style checker and you'll probably learn a lot of about standard python style from it

https://pypi.python.org/pypi/pyflakes

• pylint - a much more in-depth static code checker - it can often find subtle bugs in your code.

https://pypi.python.org/pypi/pylint

It also comments on style but can be really pedantic, even with very idiomatic python. It's extremely useful to catch cases where you typo'd a variable name that would not otherwise be caught until runtime and the style recommendations are useful to understand even when they can be ignored.

• coverage.py - shows you how much of the code your tests are actually covering

https://pypi.python.org/pypi/coverage

This integrates well with nosetests and can help you find code paths you may have missed with your test cases

• mocking - stub out complex bits of code so you can unit test your code

I use some mock libraries to fake out query/result patterns, so I can test my code for the case that "If mysql returns this result, am I doing the right thing?"

I use mocker for this in some of the holland tests:

https://pypi.python.org/pypi/mocker

For example, I might write a set of mocks that return different output for SHOW SLAVE STATUS and make sure test_slave does the right thing with that output, so I don't have to setup a full mysql replication slave to run my unit tests

Of course, with mock you still want to test against the real thing as well, but for unit tests mock objects can be really useful.