编写一个python脚本把rds慢日志发送到企业微信群

前提工作

安装aliyuncli及配置凭证

安装文档：https://help.aliyun.com/document_detail/121541.html?spm=a2c4g.29991.0.0.10dd233dbo8y7Q

配置凭证：https://help.aliyun.com/document_detail/121259.html?spm=a2c4g.121193.0.0.67b31952OdcYEw

其他功能操作：https://help.aliyun.com/document_detail/121198.html?spm=a2c4g.110289.0.0.ca1f42ec8jM3Hn

安装aliyuncli略

创建ram用户配置权限

 应只需云数据库rds权限即可，具体提阿里云工单提问，这块没做测试

配置凭证演示

aliyun configure set --profile default --access-key-id 你的key --access-key-secret 你的secert --region cn-shenzhen --language en

列出概要配置

[root@prometheus monitor-script]# aliyun configure list

Profile   | Credential         | Valid   | Region           | Language

--------- | ------------------ | ------- | ---------------- | --------

default * | AK:***s79          | Valid   | cn-shenzhen      | en

aliyun命令行工具列出慢查询语句

aliyun rds DescribeSlowLogs --DBInstanceId 你的实例ID --StartTime 2023-04-26Z --EndTime 2023-04-26Z

脚本内容如下

 需要修改的地方： 你的RDS实例ID，你的企业微信Key

 配一个定时任务。每天凌晨十点执行脚本获取昨天所有的慢日志

[root@prometheus monitor-script]# cat mysql-slow-monitor.py    
#!/usr/bin/env python3.6
# coding: utf-8

import subprocess
import requests
import json
import datetime
import time

# 定义aliyun rds命令和重试次数
command_base = ['aliyun', 'rds', 'DescribeSlowLogs', '--DBInstanceId']
retry_count = 5

# 定义Webhook URL
webhook_url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key'
    
# 定义要查询的DBInstanceId列表及其对应的实例ID
db_instance_ids = {
    '你的RDS实例ID1': 'C端RDS实例',
    '你的RDS实例ID2': 'B端RDS实例'
}

for db_instance_id in db_instance_ids:
    # 获取昨天的日期
    yesterday = datetime.date.today() - datetime.timedelta(days=1)

    # 将日期格式化为字符串
    start_time = yesterday.strftime('%Y-%m-%d') + 'Z'
    end_time = yesterday.strftime('%Y-%m-%d') + 'Z'

    # 构建aliyun rds命令
    command = command_base + [db_instance_id, '--StartTime', start_time, '--EndTime', end_time]

    # 调用aliyun rds命令获取慢查询日志
    try:
        output = subprocess.check_output(command)
    except Exception as e:
        print(f"Error occurred: {e}")
        for i in range(retry_count - 1):
            print(f"Retrying in 2 seconds ({i+1}/{retry_count})")
            time.sleep(2)
            try:
                # 重试调用aliyun rds命令获取慢查询日志
                output = subprocess.check_output(command)
                break
            except Exception as e:
                print(f"Error occurred: {e}")
        else:
            print("Maximum retries exceeded. Exiting...")
            exit()
            
	# 解析输出并发送数据到Webhook
    payloads = []

    for line in output.decode().split('\n'):
        if 'CreateTime' in line:
            createtime = line.split(': ')[1]
        elif 'DBName' in line:
            dbname = line.split(': ')[1]
        elif 'MaxExecutionTime' in line:
            maxexectime = line.split(': ')[1]
        elif 'MySQLTotalExecutionCounts' in line:
            totalexecutioncounts = line.split(': ')[1]
        elif 'MySQLTotalExecutionTimes' in line:
            totalexecutiontimes = line.split(': ')[1]
        elif 'ParseMaxRowCount' in line:
            parsemaxrowcount = line.split(': ')[1]
        elif 'ParseTotalRowCounts' in line:
            parsetotalrowcounts = line.split(': ')[1]
        elif 'ReturnMaxRowCount' in line:
            returnmaxrowcount = line.split(': ')[1]
        elif 'ReturnTotalRowCounts' in line:
            returntotalrowcounts = line.split(': ')[1]
        elif 'SQLText' in line:
            sqltext = line.split(': ')[1].strip()
            payload = {
                'msgtype': 'text',
                'text': {
                    'content': f"[{createtime}] 实例：{db_instance_ids[db_instance_id]}，数据库：{dbname}，总执行次数：{totalexecutioncounts}，总执行时长：{totalexecutiontimes}，最长执行时长{maxexectime}，最大解析行数：{parsemaxrowcount}，解析总行数：{parsetotalrowcounts}，返回最长行数：{returnmaxrowcount}，返回总行数：{returntotalrowcounts}，SQL语句为：{sqltext}"
                }
            }
            payloads.append(payload)

    for payload in payloads:
        response = requests.post(webhook_url, json=payload)

日志如下

告警如下