好的，下面是针对如何构建一个用于监控 Docker 中 MySQL 慢查询的 Python 脚本的具体指导。该脚本将在检测到慢查询时将结果发送至企业微信，并且通知中会包含关键的信息如慢查询的时间、SQL语句及其执行时间。 ### 准备工作 #### 安装依赖项 确保安装了必要的 Python 包： - `mysql-connector-python`: 与 MySQL 数据库交互。 - `requests`: 用于向企业微信 WebHook 接口发起请求。 使用 pip 安装这两个包： ```bash pip install mysql-connector-python requests ``` #### 设置 MySQL 慢查询日志 在启动 MySQL Docker 容器时通过环境变量或配置文件启用慢查询日志，并设置合理的阈值（例如超过1秒）。可以在 docker-compose.yml 文件中添加如下配置片段： ```yaml version: '3.8' services: mysql-server: image: mysql:latest container_name: mysql-container ports: - "3306:3306" environment: MYSQL_ROOT_PASSWORD: your_root_password_here MYSQL_DATABASE: testdb command: ["mysqld", "--slow-query-log=ON", "--long_query_time=1"] volumes: - ./data:/var/lib/mysql ``` 这样就启用了慢查询日志并将长查询定义为大于等于一秒的情况。 ### 编写监控脚本 我们将编写一段简单的 Python 程序来定期轮询 MySQL 并查找最新的慢查询记录。一旦发现新的慢查询，它就会把相关信息发布给预设的企业微信群聊。 以下是完整的代码示例： ```python import time import logging from datetime import datetime import mysql.connector as mc import requests # 初始化 logger logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class MysqlSlowQueryMonitor: def __init__(self, db_config, wecom_webhook_url): self.db_conn = None self.last_seen_timestamp = None self.wecom_webhook_url = wecom_webhook_url # 创建数据库连接 try: self.db_conn = mc.connect(**db_config) cursor = self.db_conn.cursor(dictionary=True) cursor.execute("SELECT NOW()") row = cursor.fetchone() initial_now = row['NOW()'] self.last_seen_timestamp = initial_now.strftime('%Y-%m-%d %H:%M:%S') except Exception as err: logger.error(f"初始化失败: {err}") raise def get_latest_slow_queries(self): """获取自 last_seen_timestamp 后所有的慢查询""" queries = [] if not self.db_conn.is_connected(): self.db_conn.reconnect(attempts=5, delay=2) cursor = self.db_conn.cursor(dictionary=True) try: cursor.execute(""" SELECT query_time, sql_text, count_star FROM INFORMATION_SCHEMA.SLOW_LOG WHERE start_time > %s; """, (self.last_seen_timestamp,)) rows = cursor.fetchall() for row in rows: timestamp = row["query_time"].strftime("%Y-%m-%d %H:%M:%S") queries.append({ "timestamp": timestamp, "sql": row["sql_text"], "count": row["count_star"] }) current_max_ts = max([row["query_time"] for row in rows]) \ if len(rows)>0 else None if current_max_ts is not None: self.last_seen_timestamp = current_max_ts.strftime('%Y-%m-%d %H:%M:%S') except Exception as err: logger.error(f"查询过程中出现问题: {err}") return queries def send_to_wecom(self, slow_query_info): message_template = "[MYSQL SLOW QUERY]\n" +\ "- 时间: {}\n"+\ "- SQL: {}\n"+\ "- 执行次数: {}" formatted_message = "\n".join([ "@所有人\n" + message_template.format(q['timestamp'], q['sql'].strip(), q['count']) for q in slow_query_info]) payload = { "msgtype": "text", "text": { "mentioned_mobile_list":["@all"], # 或者指定特定用户的手机号码列表 "content": formatted_message } } resp = requests.post(self.wecom_webhook_url, json=payload).json() success_status_codes = [200] if resp.get('errmsg') != 'ok' or resp.get('errcode') not in success_status_codes: logger.warning("发送通知失败") else: logger.info("成功发送了一组慢查询警告") def main(): DATABASE_CONFIG = { 'host': 'localhost', 'user': 'root', 'password': '<Your Password>', 'database': '<Database Name>' } WECOM_WEBHOOK_URL = "<Your WeCom Webhook URL>" monitor = MysqlSlowQueryMonitor(DATABASE_CONFIG, WECOM_WEBHOOK_URL) CHECK_INTERVAL_SECONDS = 60 # 设定每次检查之间的间隔时间为一分钟 while True: latest_slow_queries = monitor.get_latest_slow_queries() if latest_slow_queries: logger.info(f"Detected {len(latest_slow_queries)} new slow queries.") monitor.send_to_wecom(latest_slow_queries) else: logger.debug("No new slow queries detected.") time.sleep(CHECK_INTERVAL_SECONDS) if __name__ == "__main__": main() ``` 这个版本实现了以下几个特点： - **自动重连**：如果失去与 MySQL 的连接，则尝试重新建立连接最多五次。 - **持续监听**：无限循环每分钟一次地去检索新增加的慢查询条目，并及时报告出来。 - **精准定位**：利用 `last_seen_timestamp` 来区分哪些是新产生的慢查询，避免重复推送之前已经处理过的记录。 - **异常处理**：增加了基本的日志输出以及对各种潜在错误情形的支持。 请替换掉 `<Your Password>` 和 `<Database Name>` 字段的内容为自己真实的密码和目标数据库名称，同样也需填入有效的企业微信机器人WebHook地址 (`<Your WeCom Webhook URL>`) ，然后就可以直接运行这段代码啦！ 希望这对您有所帮助！如果您有任何疑问或者需要进一步的帮助，请随时告诉我。