第305集企业级数据库监控架构实战：连接数管理、慢查询分析与索引优化、智能监控与性能调优解决方案

#!/usr/bin/env python3
# slow-query-analyzer.py
# 功能：自定义慢查询分析工具，提供深度分析和优化建议

import re
import json
import logging
from collections import defaultdict
from datetime import datetime, timedelta
from typing import List, Dict, Any

class SlowQueryAnalyzer:
    """慢查询分析器"""
    
    def __init__(self, log_file: str):
        self.log_file = log_file
        self.queries = []
        
        # 🔍 正则表达式模式
        self.patterns = {
            'timestamp': r'Time: (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z)',
            'query_time': r'Query_time: ([\d.]+)',
            'lock_time': r'Lock_time: ([\d.]+)',
            'rows_sent': r'Rows_sent: (\d+)',
            'rows_examined': r'Rows_examined: (\d+)',
            'user_host': r'User@Host: (\w+)@(\w+)',
            'sql': r'SELECT.*?(?=\n# Time:|\n# User@Host:|\Z)'
        }
        
        # 📊 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)

    def parse_log(self) -> None:
        """解析慢查询日志"""
        self.logger.info(f"开始解析慢查询日志: {self.log_file}")
        
        try:
            with open(self.log_file, 'r', encoding='utf-8') as f:
                content = f.read()

            # 🔄 分割查询块
            queries = content.split('# Time:')[1:]
            self.logger.info(f"发现 {len(queries)} 个查询块")

            for i, query_block in enumerate(queries):
                query_data = self.parse_query_block(query_block)
                if query_data:
                    self.queries.append(query_data)
                    
            self.logger.info(f"成功解析 {len(self.queries)} 个慢查询")
            
        except Exception as e:
            self.logger.error(f"解析日志文件失败: {str(e)}")
            raise

    def parse_query_block(self, block: str) -> Dict[str, Any]:
        """解析单个查询块"""
        query_data = {}

        for key, pattern in self.patterns.items():
            match = re.search(pattern, block, re.DOTALL)
            if match:
                if key == 'sql':
                    query_data[key] = match.group(0).strip()
                elif key == 'user_host':
                    query_data['user'] = match.group(1)
                    query_data['host'] = match.group(2)
                else:
                    query_data[key] = match.group(1)

        return query_data if len(query_data) > 3 else None

    def analyze_queries(self) -> Dict[str, Any]:
        """分析查询性能"""
        self.logger.info("开始分析查询性能")
        
        analysis = {
            'total_queries': len(self.queries),
            'avg_query_time': 0,
            'max_query_time': 0,
            'slowest_queries': [],
            'most_frequent_queries': [],
            'queries_without_index': [],
            'performance_issues': []
        }

        if not self.queries:
            return analysis

        # 📊 计算平均查询时间
        total_time = sum(float(q.get('query_time', 0)) for q in self.queries)
        analysis['avg_query_time'] = total_time / len(self.queries)

        # 🐌 找出最慢的查询
        sorted_queries = sorted(
            self.queries, 
            key=lambda x: float(x.get('query_time', 0)), 
            reverse=True
        )
        analysis['slowest_queries'] = sorted_queries[:10]
        analysis['max_query_time'] = float(sorted_queries[0].get('query_time', 0))

        # 🔄 统计查询频率
        query_counts = defaultdict(int)
        for query in self.queries:
            sql = query.get('sql', '')
            normalized_sql = self.normalize_sql(sql)
            query_counts[normalized_sql] += 1

        analysis['most_frequent_queries'] = sorted(
            query_counts.items(), 
            key=lambda x: x[1], 
            reverse=True
        )[:10]

        # 🚫 找出没有使用索引的查询
        for query in self.queries:
            rows_examined = int(query.get('rows_examined', 0))
            rows_sent = int(query.get('rows_sent', 0))
            if rows_examined > rows_sent * 10:  # 扫描行数远大于返回行数
                analysis['queries_without_index'].append(query)

        # ⚠️ 识别性能问题
        analysis['performance_issues'] = self.identify_performance_issues()

        return analysis

    def identify_performance_issues(self) -> List[Dict[str, Any]]:
        """识别性能问题"""
        issues = []
        
        for query in self.queries:
            query_time = float(query.get('query_time', 0))
            rows_examined = int(query.get('rows_examined', 0))
            rows_sent = int(query.get('rows_sent', 0))
            
            # 🐌 超慢查询
            if query_time > 10:
                issues.append({
                    'type': 'slow_query',
                    'severity': 'critical',
                    'query': query.get('sql', '')[:100] + '...',
                    'query_time': query_time,
                    'recommendation': '优化查询逻辑或添加索引'
                })
            
            # 📊 低效查询（扫描行数过多）
            if rows_examined > rows_sent * 100:
                issues.append({
                    'type': 'inefficient_query',
                    'severity': 'warning',
                    'query': query.get('sql', '')[:100] + '...',
                    'rows_examined': rows_examined,
                    'rows_sent': rows_sent,
                    'recommendation': '添加合适的索引或优化WHERE条件'
                })
        
        return issues

    def normalize_sql(self, sql: str) -> str:
        """标准化SQL语句"""
        # 移除多余空格
        sql = re.sub(r'\s+', ' ', sql.strip())
        # 移除具体值，用占位符替换
        sql = re.sub(r'\d+', '?', sql)
        sql = re.sub(r"'[^']*'", "'?'", sql)
        sql = re.sub(r'"[^"]*"', '"?"', sql)
        return sql

    def generate_report(self) -> str:
        """生成分析报告"""
        analysis = self.analyze_queries()

        report = f"""
# 📊 慢查询分析报告

## 📈 概览统计
- **总查询数**: {analysis['total_queries']}
- **平均查询时间**: {analysis['avg_query_time']:.2f}秒
- **最大查询时间**: {analysis['max_query_time']:.2f}秒

## 🐌 最慢的10个查询
"""

        for i, query in enumerate(analysis['slowest_queries'], 1):
            report += f"""
### {i}. 查询时间: {query.get('query_time', 'N/A')}秒
```sql
{query.get('sql', 'N/A')[:200]}...

#!/usr/bin/env python3
# index-analyzer.py
# 功能：自动分析数据库索引效率，提供优化建议

import mysql.connector
import json
import logging
from collections import defaultdict
from typing import List, Dict, Any
from datetime import datetime

class IndexAnalyzer:
    """索引分析器"""
    
    def __init__(self, host: str, user: str, password: str, database: str):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        
        # 📊 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)

    def connect(self) -> mysql.connector.connection.MySQLConnection:
        """建立数据库连接"""
        return mysql.connector.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database
        )

    def analyze_index_usage(self) -> List[Dict[str, Any]]:
        """分析索引使用情况"""
        self.logger.info("开始分析索引使用情况")
        
        conn = self.connect()
        cursor = conn.cursor(dictionary=True)
        
        try:
            query = """
            SELECT 
                OBJECT_SCHEMA,
                OBJECT_NAME,
                INDEX_NAME,
                COUNT_FETCH,
                COUNT_INSERT,
                COUNT_UPDATE,
                COUNT_DELETE,
                COUNT_FETCH + COUNT_INSERT + COUNT_UPDATE + COUNT_DELETE as TOTAL_USAGE
            FROM performance_schema.table_io_waits_summary_by_index_usage
            WHERE OBJECT_SCHEMA = %s
            ORDER BY TOTAL_USAGE DESC
            """
            
            cursor.execute(query, (self.database,))
            return cursor.fetchall()
            
        finally:
            cursor.close()
            conn.close()

    def analyze_unused_indexes(self) -> List[Dict[str, Any]]:
        """分析未使用的索引"""
        self.logger.info("分析未使用的索引")
        
        conn = self.connect()
        cursor = conn.cursor(dictionary=True)
        
        try:
            query = """
            SELECT 
                OBJECT_SCHEMA,
                OBJECT_NAME,
                INDEX_NAME,
                COUNT_FETCH,
                COUNT_INSERT,
                COUNT_UPDATE,
                COUNT_DELETE
            FROM performance_schema.table_io_waits_summary_by_index_usage
            WHERE OBJECT_SCHEMA = %s
            AND COUNT_FETCH = 0
            AND COUNT_INSERT = 0
            AND COUNT_UPDATE = 0
            AND COUNT_DELETE = 0
            AND INDEX_NAME IS NOT NULL
            AND INDEX_NAME != 'PRIMARY'
            """
            
            cursor.execute(query, (self.database,))
            return cursor.fetchall()
            
        finally:
            cursor.close()
            conn.close()

    def analyze_index_cardinality(self) -> List[Dict[str, Any]]:
        """分析索引基数"""
        self.logger.info("分析索引基数")
        
        conn = self.connect()
        cursor = conn.cursor(dictionary=True)
        
        try:
            query = """
            SELECT 
                TABLE_SCHEMA,
                TABLE_NAME,
                INDEX_NAME,
                CARDINALITY,
                TABLE_ROWS,
                CASE 
                    WHEN TABLE_ROWS > 0 THEN CARDINALITY / TABLE_ROWS 
                    ELSE 0 
                END as SELECTIVITY
            FROM INFORMATION_SCHEMA.STATISTICS s
            JOIN INFORMATION_SCHEMA.TABLES t 
            ON s.TABLE_SCHEMA = t.TABLE_SCHEMA 
            AND s.TABLE_NAME = t.TABLE_NAME
            WHERE s.TABLE_SCHEMA = %s
            AND t.TABLE_ROWS > 0
            ORDER BY SELECTIVITY DESC
            """
            
            cursor.execute(query, (self.database,))
            return cursor.fetchall()
            
        finally:
            cursor.close()
            conn.close()

    def analyze_duplicate_indexes(self) -> List[Dict[str, Any]]:
        """分析重复索引"""
        self.logger.info("分析重复索引")
        
        conn = self.connect()
        cursor = conn.cursor(dictionary=True)
        
        try:
            query = """
            SELECT 
                TABLE_SCHEMA,
                TABLE_NAME,
                GROUP_CONCAT(INDEX_NAME) as INDEX_NAMES,
                GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX) as COLUMNS,
                COUNT(*) as INDEX_COUNT
            FROM INFORMATION_SCHEMA.STATISTICS
            WHERE TABLE_SCHEMA = %s
            AND INDEX_NAME != 'PRIMARY'
            GROUP BY TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME
            HAVING COUNT(*) > 1
            """
            
            cursor.execute(query, (self.database,))
            return cursor.fetchall()
            
        finally:
            cursor.close()
            conn.close()

    def generate_optimization_recommendations(self) -> Dict[str, Any]:
        """生成优化建议"""
        self.logger.info("生成索引优化建议")
        
        recommendations = {
            'unused_indexes': [],
            'low_selectivity_indexes': [],
            'duplicate_indexes': [],
            'missing_indexes': [],
            'summary': {
                'total_recommendations': 0,
                'critical_issues': 0,
                'warning_issues': 0
            }
        }

        # 🚫 未使用的索引
        unused_indexes = self.analyze_unused_indexes()
        for index in unused_indexes:
            recommendations['unused_indexes'].append({
                'table': f"{index['OBJECT_SCHEMA']}.{index['OBJECT_NAME']}",
                'index': index['INDEX_NAME'],
                'severity': 'warning',
                'recommendation': '考虑删除此未使用的索引以节省存储空间和提升写入性能',
                'action': f"DROP INDEX {index['INDEX_NAME']} ON {index['OBJECT_SCHEMA']}.{index['OBJECT_NAME']}"
            })

        # 📊 低选择性索引
        cardinality_data = self.analyze_index_cardinality()
        for index in cardinality_data:
            if index['SELECTIVITY'] < 0.1:  # 选择性低于10%
                recommendations['low_selectivity_indexes'].append({
                    'table': f"{index['TABLE_SCHEMA']}.{index['TABLE_NAME']}",
                    'index': index['INDEX_NAME'],
                    'selectivity': f"{index['SELECTIVITY']:.2%}",
                    'severity': 'warning',
                    'recommendation': '索引选择性过低，可能影响查询性能',
                    'action': '考虑优化索引列的顺序或添加更多选择性高的列'
                })

        # 🔄 重复索引
        duplicate_indexes = self.analyze_duplicate_indexes()
        for index in duplicate_indexes:
            recommendations['duplicate_indexes'].append({
                'table': f"{index['TABLE_SCHEMA']}.{index['TABLE_NAME']}",
                'indexes': index['INDEX_NAMES'],
                'columns': index['COLUMNS'],
                'severity': 'critical',
                'recommendation': '存在重复索引，浪费存储空间',
                'action': f"保留一个索引，删除其他重复索引"
            })

        # 📈 统计信息
        recommendations['summary']['total_recommendations'] = (
            len(recommendations['unused_indexes']) +
            len(recommendations['low_selectivity_indexes']) +
            len(recommendations['duplicate_indexes'])
        )
        recommendations['summary']['critical_issues'] = len(recommendations['duplicate_indexes'])
        recommendations['summary']['warning_issues'] = (
            len(recommendations['unused_indexes']) +
            len(recommendations['low_selectivity_indexes'])
        )

        return recommendations

    def generate_report(self) -> str:
        """生成分析报告"""
        recommendations = self.generate_optimization_recommendations()
        
        report = f"""
# 📊 索引优化分析报告

## 📈 概览统计
- **总建议数**: {recommendations['summary']['total_recommendations']}
- **严重问题**: {recommendations['summary']['critical_issues']}
- **警告问题**: {recommendations['summary']['warning_issues']}

## 🚫 未使用的索引
"""
        
        for index in recommendations['unused_indexes']:
            report += f"""
### {index['table']}.{index['index']}
- **严重程度**: {index['severity'].upper()}
- **建议**: {index['recommendation']}
- **操作**: `{index['action']}`
"""

        report += "\n## 📊 低选择性索引\n"
        for index in recommendations['low_selectivity_indexes']:
            report += f"""
### {index['table']}.{index['index']}
- **选择性**: {index['selectivity']}
- **严重程度**: {index['severity'].upper()}
- **建议**: {index['recommendation']}
- **操作**: {index['action']}
"""

        report += "\n## 🔄 重复索引\n"
        for index in recommendations['duplicate_indexes']:
            report += f"""
### {index['table']}
- **重复索引**: {index['indexes']}
- **列**: {index['columns']}
- **严重程度**: {index['severity'].upper()}
- **建议**: {index['recommendation']}
- **操作**: {index['action']}
"""

        return report

    def close(self):
        """关闭连接"""
        pass  # 连接在方法中已经关闭

if __name__ == "__main__":
    analyzer = IndexAnalyzer('localhost', 'root', 'password', 'your_database')
    
    # 生成分析报告
    recommendations = analyzer.generate_optimization_recommendations()
    report = analyzer.generate_report()
    
    print("=== 索引优化建议 ===")
    print(json.dumps(recommendations, indent=2, ensure_ascii=False))
    print("\n=== 详细报告 ===")
    print(report)
    
    analyzer.close()

#!/usr/bin/env python3
# auto-index-optimizer.py

import mysql.connector
import schedule
import time
import logging

class AutoIndexOptimizer:
    def __init__(self, host, user, password, database):
        self.host = host
        self.user = user
        self.password = password
        self.database = database

        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('index-optimizer.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

    def connect(self):
        """建立数据库连接"""
        return mysql.connector.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database
        )

    def analyze_and_optimize(self):
        """分析并优化索引"""
        conn = self.connect()
        cursor = conn.cursor(dictionary=True)

        try:
            # 获取需要优化的表
            query = """
            SELECT 
                TABLE_NAME,
                TABLE_ROWS,
                DATA_LENGTH,
                INDEX_LENGTH
            FROM INFORMATION_SCHEMA.TABLES
            WHERE TABLE_SCHEMA = %s
            AND TABLE_ROWS > 1000
            ORDER BY TABLE_ROWS DESC
            """

            cursor.execute(query, (self.database,))
            tables = cursor.fetchall()

            for table in tables:
                self.optimize_table(cursor, table['TABLE_NAME'])

        finally:
            cursor.close()
            conn.close()

    def optimize_table(self, cursor, table_name):
        """优化单个表"""
        try:
            self.logger.info(f"开始优化表: {table_name}")

            # 分析表
            cursor.execute(f"ANALYZE TABLE {table_name}")
            self.logger.info(f"表 {table_name} 分析完成")

            # 检查表
            cursor.execute(f"CHECK TABLE {table_name}")
            result = cursor.fetchone()
            if result['Msg_text'] != 'OK':
                self.logger.warning(f"表 {table_name} 检查发现问题: {result['Msg_text']}")

            # 优化表
            cursor.execute(f"OPTIMIZE TABLE {table_name}")
            self.logger.info(f"表 {table_name} 优化完成")

        except Exception as e:
            self.logger.error(f"优化表 {table_name} 时出错: {str(e)}")

    def start_scheduler(self):
        """启动定时任务"""
        # 每天凌晨2点执行优化
        schedule.every().day.at("02:00").do(self.analyze_and_optimize)

        # 每周日凌晨3点执行深度优化
        schedule.every().sunday.at("03:00").do(self.deep_optimize)

        self.logger.info("索引优化调度器已启动")

        while True:
            schedule.run_pending()
            time.sleep(60)

    def deep_optimize(self):
        """深度优化"""
        self.logger.info("开始深度优化")

        conn = self.connect()
        cursor = conn.cursor(dictionary=True)

        try:
            # 重建所有索引
            query = """
            SELECT DISTINCT TABLE_NAME
            FROM INFORMATION_SCHEMA.STATISTICS
            WHERE TABLE_SCHEMA = %s
            AND INDEX_NAME != 'PRIMARY'
            """

            cursor.execute(query, (self.database,))
            tables = cursor.fetchall()

            for table in tables:
                table_name = table['TABLE_NAME']
                self.logger.info(f"重建表 {table_name} 的索引")

                # 获取所有非主键索引
                index_query = """
                SELECT INDEX_NAME
                FROM INFORMATION_SCHEMA.STATISTICS
                WHERE TABLE_SCHEMA = %s
                AND TABLE_NAME = %s
                AND INDEX_NAME != 'PRIMARY'
                """

                cursor.execute(index_query, (self.database, table_name))
                indexes = cursor.fetchall()

                for index in indexes:
                    index_name = index['INDEX_NAME']
                    try:
                        cursor.execute(f"ALTER TABLE {table_name} DROP INDEX {index_name}")
                        cursor.execute(f"ALTER TABLE {table_name} ADD INDEX {index_name}")
                        self.logger.info(f"重建索引 {index_name} 完成")
                    except Exception as e:
                        self.logger.error(f"重建索引 {index_name} 失败: {str(e)}")

        finally:
            cursor.close()
            conn.close()

        self.logger.info("深度优化完成")

if __name__ == "__main__":
    optimizer = AutoIndexOptimizer('localhost', 'root', 'password', 'your_database')
    optimizer.start_scheduler()

#!/usr/bin/env python3
# performance-monitor.py

import mysql.connector
import time
import json
import logging
from datetime import datetime

class PerformanceMonitor:
    def __init__(self, host, user, password, database):
        self.host = host
        self.user = user
        self.password = password
        self.database = database

        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def connect(self):
        """建立数据库连接"""
        return mysql.connector.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database
        )

    def collect_metrics(self):
        """收集性能指标"""
        conn = self.connect()
        cursor = conn.cursor(dictionary=True)

        metrics = {
            'timestamp': datetime.now().isoformat(),
            'connections': {},
            'innodb': {},
            'query_cache': {},
            'threads': {},
            'slow_queries': {}
        }

        try:
            # 连接数指标
            cursor.execute("SHOW STATUS LIKE 'Threads_connected'")
            metrics['connections']['threads_connected'] = cursor.fetchone()['Value']

            cursor.execute("SHOW STATUS LIKE 'Threads_running'")
            metrics['connections']['threads_running'] = cursor.fetchone()['Value']

            cursor.execute("SHOW STATUS LIKE 'Max_used_connections'")
            metrics['connections']['max_used_connections'] = cursor.fetchone()['Value']

            # InnoDB指标
            cursor.execute("SHOW STATUS LIKE 'Innodb_buffer_pool_hit_rate'")
            result = cursor.fetchone()
            if result:
                metrics['innodb']['buffer_pool_hit_rate'] = result['Value']

            cursor.execute("SHOW STATUS LIKE 'Innodb_log_waits'")
            metrics['innodb']['log_waits'] = cursor.fetchone()['Value']

            # 查询缓存指标
            cursor.execute("SHOW STATUS LIKE 'Qcache_hits'")
            metrics['query_cache']['hits'] = cursor.fetchone()['Value']

            cursor.execute("SHOW STATUS LIKE 'Qcache_inserts'")
            metrics['query_cache']['inserts'] = cursor.fetchone()['Value']

            # 慢查询指标
            cursor.execute("SHOW STATUS LIKE 'Slow_queries'")
            metrics['slow_queries']['count'] = cursor.fetchone()['Value']

        finally:
            cursor.close()
            conn.close()

        return metrics

    def analyze_performance(self, metrics):
        """分析性能指标"""
        issues = []

        # 检查连接数
        threads_connected = int(metrics['connections']['threads_connected'])
        threads_running = int(metrics['connections']['threads_running'])

        if threads_connected > 800:  # 假设最大连接数为1000
            issues.append({
                'type': 'high_connections',
                'severity': 'warning',
                'message': f'连接数过高: {threads_connected}',
                'recommendation': '检查连接池配置或增加最大连接数'
            })

        if threads_running > 50:
            issues.append({
                'type': 'high_running_threads',
                'severity': 'critical',
                'message': f'运行线程数过高: {threads_running}',
                'recommendation': '检查慢查询或优化数据库性能'
            })

        # 检查InnoDB缓冲池命中率
        if 'buffer_pool_hit_rate' in metrics['innodb']:
            hit_rate = float(metrics['innodb']['buffer_pool_hit_rate'])
            if hit_rate < 95:
                issues.append({
                    'type': 'low_buffer_pool_hit_rate',
                    'severity': 'warning',
                    'message': f'缓冲池命中率过低: {hit_rate}%',
                    'recommendation': '增加innodb_buffer_pool_size'
                })

        return issues

    def start_monitoring(self, interval=60):
        """开始监控"""
        self.logger.info("开始性能监控")

        while True:
            try:
                metrics = self.collect_metrics()
                issues = self.analyze_performance(metrics)

                if issues:
                    self.logger.warning(f"发现性能问题: {len(issues)}个")
                    for issue in issues:
                        self.logger.warning(f"{issue['severity'].upper()}: {issue['message']}")
                        self.logger.info(f"建议: {issue['recommendation']}")

                # 保存指标到文件
                with open(f'performance_metrics_{datetime.now().strftime("%Y%m%d")}.json', 'a') as f:
                    f.write(json.dumps(metrics) + '\n')

                time.sleep(interval)

            except Exception as e:
                self.logger.error(f"监控过程中出错: {str(e)}")
                time.sleep(interval)

if __name__ == "__main__":
    monitor = PerformanceMonitor('localhost', 'root', 'password', 'your_database')
    monitor.start_monitoring()