使用 Python 监控 MySQL 慢查询日志

生产环境里有时会出现一些比较耗时的查询，我们希望能在第一时间知道具体是哪些查询拖慢数据库。当检测到有慢查询日至产生的时候，就发送邮件通知管理员。因此我的一个想法就是去轮询MySQL慢查询日志。当然，这么做的前提是慢查询日志是打开的。当时花了 2 个小时实现的这个功能，现在看起来还是比较有用的，现在把我饿方法贴出来，如果谁有更好的方法，可一起交流一下。

这个后台监控脚本分为 3 个模块来实现：

• 日至检测模块
• 邮件发送模块
• 计时模块，同时把上面两个模块整合在一起

先看日至检测模块：

# 传入文件的路径，以及上次读区日志的位置
# 返回一条慢SQL和本次读取日志的位置
def read_file(file_path, pre_pos=0):
    next_pos = pre_pos + 1
    cur_pos = 0
    slow_sql = ''
    with open(file_path, 'r') as f:
        for line in f:
            if re.match('^#', line) and re.search('Time', line):
                cur_pos = cur_pos + 1
                if cur_pos == next_pos:
                    slow_sql = slow_sql + line
                    for line in f:
                        if re.match('^#', line) and re.search('Time', line):
                            cur_pos = cur_pos + 1
                            slow_sql = slow_sql + line
                        else:
                            slow_sql = slow_sql + line
                    return slow_sql, cur_pos
        return None

# 统计慢SQL的行数
def sql_count(file_path):
    row_cnt = 0
    with open(file_path, 'r') as f:
        for line in f:
            if re.match('^#', line) and re.search('Time', line):
                row_cnt = row_cnt + 1
        return row_cnt

接下来是邮件发送模块，这里我使用了第三方封装好的邮件发送的模块，系统原生的比较难用：

from envelopes import Envelope

def send_mail(mail_list, subject, mail_body):
    envelope = Envelope(
        from_addr=('abc@abc.com', 'MySQL数据库监控'),
        to_addr=(mail_list),
        cc_addr=(['abc@abc.com']),
        subject=subject,
        text_body=mail_body
    )
    # 发送附件
    # envelope.add_attachment('/Users/light.zhang/test.jpg')
    envelope.send('smtp.abc.com', login='abc@abc.com', password='lumulu2tipep', tls=True)

最后一个地方是把这连个模块融合到一起，同时又计时功能：

# 主程序
def main():
    # configure the logging system
    logging.basicConfig(
        filename='app.log',
        level=logging.INFO,
        format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s %(funcName)s',
        filemode='w'
    )
    assert slow_log_path != ''
    mail_list = collect_mail_address('./mail_address.txt')
    log_pos = 0
    while True:
        return_val = rl.read_file(slow_log_path, log_pos)
        logging.info('logging')
        if return_val is not None:
            slow_sql, log_pos = return_val
            now = int(time.time())
            timeArray = time.localtime(now)
            str_time = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
            subject = 'SLOW SQL Statement %s' % str_time
            sd.send_mail(mail_list, subject, slow_sql)
            print(subject)
        time.sleep(180)

肯定有可以改进的地方，欢迎吐槽。