《MySQL 简易速速上手小册》第7章：MySQL监控和日志分析（2024 最新版）

import requests
import json
# Prometheus 查询接口
PROMETHEUS = 'http://localhost:9090/api/v1/query'
def query_prometheus(query):
    response = requests.get(PROMETHEUS, params={'query': query})
    results = response.json().get('data', {}).get('result', [])
    return results
def check_mysql_load():
    query = 'rate(mysql_global_status_questions[5m])'
    result = query_prometheus(query)
    for metric in result:
        print(f"Query Load: {metric['value'][1]}")
if __name__ == "__main__":
    check_mysql_load()

import subprocess
import re
def analyze_slow_queries(log_file_path):
    # 使用 pt-query-digest 分析慢查询日志
    process = subprocess.Popen(['pt-query-digest', log_file_path], stdout=subprocess.PIPE)
    output, error = process.communicate()
    if error:
        print(f"Error: {error}")
    else:
        print(output.decode())
if __name__ == "__main__":
    analyze_slow_queries('/var/lib/mysql/your-slow.log')

import smtplib
from email.mime.text import MIMEText
def send_email(subject, message):
    sender = 'your_email@example.com'
    receivers = ['receiver@example.com']
    msg = MIMEText(message)
    msg['Subject'] = subject
    msg['From'] = sender
    msg['To'] = ", ".join(receivers)
    try:
        smtpObj = smtplib.SMTP('localhost')
        smtpObj.sendmail(sender, receivers, msg.as_string())         
        print("Successfully sent email")
    except smtplib.SMTPException:
        print("Error: unable to send email")
# 在这里调用你的监控函数，使用 send_email 发送警报

import re
def parse_slow_log(log_file_path):
    with open(log_file_path, 'r') as file:
        slow_queries = file.read()
    # 使用正则表达式匹配查询和查询时间
    pattern = re.compile(r'Query_time: (\d+\.\d+) .*\n# User@Host: .*?\n(# Query_time: .*?\n)?(SET timestamp=.*;\n)?(use \w+;\n)?(.*)')
    matches = pattern.findall(slow_queries)
    
    for match in matches:
        print(f"Query Time: {match[0]}, Query: {match[4].strip()}\n")
parse_slow_log('/var/lib/mysql/slow.log')

import time
import smtplib
from email.mime.text import MIMEText
def send_email(subject, message):
    msg = MIMEText(message)
    msg['Subject'] = subject
    # 配置你的发件人和收件人信息
    msg['From'] = 'sender@example.com'
    msg['To'] = 'receiver@example.com'
    with smtplib.SMTP('localhost') as server:
        server.send_message(msg)
def monitor_error_log(log_file_path, last_pos):
    with open(log_file_path, 'r') as file:
        file.seek(last_pos)
        new_logs = file.read()
        if new_logs:
            send_email("MySQL Error Log Alert", new_logs)
        return file.tell()
last_position = 0
while True:
    last_position = monitor_error_log('/var/log/mysql/error.log', last_position)
    time.sleep(60)  # 每分钟检查一次

from collections import Counter
def analyze_general_log(log_file_path):
    with open(log_file_path, 'r') as file:
        queries = [line for line in file if 'Query' in line]
    query_counter = Counter(queries)
    most_common_queries = query_counter.most_common(10)
       for query, count in most_common_queries:
           print(f"Query: {query.strip()}, Count: {count}")
analyze_general_log('/var/lib/mysql/general.log')

import mysql.connector
def explain_query(query):
    conn = mysql.connector.connect(host='localhost', user='user', password='password', database='dbname')
    cursor = conn.cursor()
    cursor.execute(f"EXPLAIN {query}")
    result = cursor.fetchall()
    print("EXPLAIN Output:")
    for row in result:
        print(row)
    cursor.close()
    conn.close()
if __name__ == "__main__":
    query = "SELECT * FROM your_table WHERE some_column = 'value'"
    explain_query(query)

# 假设这是一个简化的例子，实际情况可能更复杂
def analyze_indexes(dbname):
    conn = mysql.connector.connect(host='localhost', user='user', password='password', database=dbname)
    cursor = conn.cursor()
    cursor.execute("SHOW TABLES")
    tables = cursor.fetchall()
    for (table,) in tables:
        cursor.execute(f"SHOW INDEX FROM {table}")
        indexes = cursor.fetchall()
        if not indexes:
            print(f"Table '{table}' does not have indexes, consider adding one.")
    cursor.close()
    conn.close()
analyze_indexes('your_dbname')

def adjust_innodb_buffer_pool_size(new_size):
    conn = mysql.connector.connect(host='localhost', user='root', password='yourpassword')
    cursor = conn.cursor()
    cursor.execute(f"SET GLOBAL innodb_buffer_pool_size = {new_size}")
    cursor.close()
    conn.close()
    print(f"InnoDB buffer pool size adjusted to {new_size}.")
# 调用这个函数前，请确保你理解调整 innodb_buffer_pool_size 对你的系统的影响
# adjust_innodb_buffer_pool_size(1024 * 1024 * 1024)  # 1GB