Flask 定时检测数据库执行时间长的SQL语句

背景：工作中发现所使用的阿里云数据库没有针对执行时间长的SQL语句进行及时报警策略，慢SQL是执行完之后才会出现，不符我的需求。固想自己写一个简单的定时任务进行检测。
在 PyCharm 中创建一个项目 alarm4mysql2, 里面有一个 Python package：operate

check_mysql.py 操作数据库相关的文件

# __author__: "klvchen"
# date: 2020/8/19
import pymysql
from operate import dingtalk     # 导入自己的钉钉报警模块


# 为了安全，专门创建了一个 MySQL用户 check_mysql,MySQL 语句: grant process on *.* to check_mysql@'%' identified by 'xxxxx'; flush privileges; 
# 定义数据连接信息
PY_MYSQL_CONN_DICT1 = {
    "host": 'xxxxx.mysql.rds.aliyuncs.com',
    "port": 3306,
    "user": 'check_mysql',
    "passwd": 'xxxxx'
}


# 定义数据连接信息
PY_MYSQL_CONN_DICT2 = {
    "host": '192.168.0.199',
    "port": 3306,
    "user": 'check_mysql',
    "passwd": 'xxxxx'
}


def check_mysql(MYSQL_DIS):
    r = 999
    try:
        conn = pymysql.connect(**MYSQL_DIS)
        cursor = conn.cursor()

        # 执行的 SQL 语句，查询 information_schema.PROCESSLIST 表中的 非 Sleep 和 Connect 状态并且执行时间超过 3 分钟的语句
        cursor.execute("SELECT count(1) FROM information_schema.PROCESSLIST WHERE command not in ('Sleep', 'Connect') and Time > 180;")

        r = cursor.fetchone()[0]

    except Exception as e:
        print('Exception', e)
    finally:
        cursor.close()
        conn.close()
        return r


# 控制报警次数
alarm_num = 3


# 重设报警次数
def reset_alarm_num():
    global alarm_num
    alarm_num = 3


def main(MYSQL_DIS):
    result = check_mysql(MYSQL_DIS)
    # 使用全局变量 alarm_num
    global alarm_num

    # 如果 SQL 语句查询结果不等于0 并且 报警次数大于 0 则进行报警，当一天中 alarm_num 小于等于0后，则不会报警，每天0点1分会调用 reset_alarm_num 函数进行 alarm_num 重置 
    if result != 0 and alarm_num > 0:
        # 发送钉钉报警
        msg = MYSQL_DIS.get('host')
        dingtalk.send_msg(msg)
        alarm_num = alarm_num - 1

check_mysql.py 操作数据库相关的文件

# __author__: "klvchen"
# date: 2020/8/20

import json
import requests


def send_msg(msg):
    url = 'https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxxxxxxxxxxxxxxxxx'  # 替换成自己的 access_token
    parameter = {
        "msgtype": "text",
        "text": {
            "content": "告警 -- 数据库：%s 有语句执行时间过长，请及时查看~" % msg  # 这里注意，我这边钉钉机器人配置了 告警 为关键字，需要根据自己真实情况进行替换
        },
    }
    headers = {
        'Content-Type': 'application/json'
    }
    requests.post(url, data=json.dumps(parameter), headers=headers)

app.py 文件

from flask import Flask
from operate import check_mysql
from flask_apscheduler import APScheduler


app = Flask(__name__)


@app.route('/')
def hello_world():
    return 'check mysql demo is running!'


# 任务配置类, 同时启动 3 个定时任务
class SchedulerConfig(object):
    JOBS = [
        {
            'id': 'check_mysql1',                        # 任务id
            'func': '__main__:check_mysql.main',         # 任务执行程序
            'args': (check_mysql.PY_MYSQL_CONN_DICT1,),  # 执行程序参数
            'trigger': 'interval',                       # 任务执行类型，定时器
            'seconds': 180                               # 任务执行时间，单位秒 (每3分钟调用一次)
        },
        {
            'id': 'check_mysql2',                        # 任务id
            'func': '__main__:check_mysql.main',         # 任务执行程序
            'args': (check_mysql.PY_MYSQL_CONN_DICT2,),  # 执行程序参数
            'trigger': 'interval',                       # 任务执行类型，定时器
            'seconds': 180                               # 任务执行时间，单位秒 (每3分钟调用一次)
        },
        {
            'id': 'reset_alarm_num',
            'func': '__main__:check_mysql.reset_alarm_num',  # 重设 alarm_num
            'args': None,
            'trigger': 'cron',
            'day_of_week': 'mon-sun',                        # 每天0点1分调用一次
            'hour': 0,
            'minute': 1
        }
    ]


if __name__ == '__main__':
    # app.debug = True                         # debug 模式下会导致 APScheduler 一次执行两次
    app.config.from_object(SchedulerConfig())  # 为实例化的flask引入定时任务配置
    scheduler = APScheduler()                  # 实例化APScheduler
    scheduler.init_app(app)                    # 把任务列表载入实例flask
    scheduler.start()                          # 启动任务计划

    app.run()

选择把改程序做成 docker 镜像运行

在 PyCharm 中 使用命令导出依赖
pip freeze > requirements.txt

把开发好的代码传到一个安装好 docker 的 centos 机器上

# 编辑 Dockerfile
FROM python:3.6
WORKDIR /data

COPY requirements.txt ./
RUN pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/

COPY . .

CMD ["python", "app.py"]

# 打包镜像
docker build -t flask_check_mysql:0.1 ./

# 若不想每次都安装依赖，可以打包一个已经安装好依赖的镜像，后面引用该镜像，把程序放进去即可。

启动

docker run --name check_mysql4aliyun -d flask_check_mysql:0.1