import mysql.connector from mysql.connector import Error from typing import Dict, Any, List, Tuple import logging class MySQLHealthCheck: def __init__(self, host: str, database: str, user: str, password: str, port: int = 3306): self.host = host self.database = database self.user = user self.password = password self.port = port self.connection = None self.logger = logging.getLogger(__name__) def connect(self) -> None: """建立与MySQL数据库的连接""" try: self.connection = mysql.connector.connect( host=self.host, database=self.database, user=self.user, password=self.password, port=self.port ) if self.connection.is_connected(): self.logger.info("成功连接到MySQL数据库") except Error as e: self.logger.error(f"连接MySQL数据库时出错: {e}") raise def disconnect(self) -> None: """关闭与MySQL数据库的连接""" if self.connection and self.connection.is_connected(): self.connection.close() self.logger.info("MySQL数据库连接已关闭") def execute_query(self, query: str) -> List[Tuple]: """执行SQL查询并返回结果""" try: cursor = self.connection.cursor() cursor.execute(query) result = cursor.fetchall() cursor.close() return result except Error as e: self.logger.error(f"执行查询时出错: {e}") raise def get_variable(self, variable_name: str) -> str: """获取MySQL变量的值""" query = f"SHOW VARIABLES LIKE '{variable_name}'" result = self.execute_query(query) return result[0][1] if result else None def get_status(self, status_name: str) -> str: """获取MySQL状态值""" query = f"SHOW STATUS LIKE '{status_name}'" result = self.execute_query(query) return result[0][1] if result else None def check_basic_config(self) -> Dict[str, Any]: """检查基本配置和状态""" return { "lower_case_table_names": self.get_variable("lower_case_table_names"), "port": self.get_variable("port"), "socket": self.get_variable("socket"), "skip_name_resolve": self.get_variable("skip_name_resolve"), "character_set_server": self.get_variable("character_set_server"), "collation_server": self.get_variable("collation_server"), "interactive_timeout": self.get_variable("interactive_timeout"), "wait_timeout": self.get_variable("wait_timeout"), "tx_isolation": self.get_variable("tx_isolation"), "datadir": self.get_variable("datadir"), "max_allowed_packet": self.get_variable("max_allowed_packet"), "thread_stack": self.get_variable("thread_stack"), "thread_cache_size": self.get_variable("thread_cache_size"), "table_open_cache": self.get_variable("table_open_cache"), "tmp_table_size": self.get_variable("tmp_table_size"), "max_heap_table_size": self.get_variable("max_heap_table_size"), "pid_file": self.get_variable("pid_file"), "log_error": self.get_variable("log_error"), "slow_query_log_file": self.get_variable("slow_query_log_file"), "long_query_time": self.get_variable("long_query_time"), } def check_connection_management(self) -> Dict[str, Any]: """检查连接管理""" return { "max_connections": self.get_variable("max_connections"), "max_used_connections": self.get_status("Max_used_connections"), "max_used_connections_time": self.get_status("Max_used_connections_time"), "max_connect_errors": self.get_variable("max_connect_errors"), "threads_connected": self.get_status("Threads_connected"), "threads_cached": self.get_status("Threads_cached"), "threads_created": self.get_status("Threads_created"), "threads_running": self.get_status("Threads_running"), } def check_binlog_config(self) -> Dict[str, Any]: """检查Binlog配置""" return { "log_bin": self.get_variable("log_bin"), "sync_binlog": self.get_variable("sync_binlog"), "binlog_format": self.get_variable("binlog_format"), "log_bin_basename": self.get_variable("log_bin_basename"), "log_bin_index": self.get_variable("log_bin_index"), "max_binlog_size": self.get_variable("max_binlog_size"), "expire_logs_days": self.get_variable("expire_logs_days"), "server_id": self.get_variable("server_id"), } def check_gtid_config(self) -> Dict[str, Any]: """检查GTID配置""" return { "gtid_mode": self.get_variable("gtid_mode"), "enforce_gtid_consistency": self.get_variable("enforce_gtid_consistency"), "log_slave_updates": self.get_variable("log_slave_updates"), } def check_innodb_config(self) -> Dict[str, Any]: """检查InnoDB配置""" return { "innodb_version": self.get_variable("innodb_version"), "innodb_file_per_table": self.get_variable("innodb_file_per_table"), "innodb_buffer_pool_size": self.get_variable("innodb_buffer_pool_size"), "innodb_log_file_size": self.get_variable("innodb_log_file_size"), "innodb_thread_concurrency": self.get_variable("innodb_thread_concurrency"), "innodb_max_dirty_pages_pct": self.get_variable("innodb_max_dirty_pages_pct"), "innodb_io_capacity_max": self.get_variable("innodb_io_capacity_max"), "innodb_log_buffer_size": self.get_variable("innodb_log_buffer_size"), "innodb_flush_log_at_trx_commit": self.get_variable("innodb_flush_log_at_trx_commit"), "innodb_io_capacity": self.get_variable("innodb_io_capacity"), "innodb_buffer_pool_instances": self.get_variable("innodb_buffer_pool_instances"), } def check_performance(self) -> Dict[str, Any]: """检查性能指标""" return { "slow_queries": self.get_status("Slow_queries"), "innodb_rows_read": self.get_status("Innodb_rows_read"), "innodb_rows_inserted": self.get_status("Innodb_rows_inserted"), "innodb_rows_updated": self.get_status("Innodb_rows_updated"), "innodb_rows_deleted": self.get_status("Innodb_rows_deleted"), } def run_health_check(self) -> Dict[str, Any]: """运行完整的健康检查""" try: self.connect() health_check_result = { "basic_config": self.check_basic_config(), "connection_management": self.check_connection_management(), "binlog_config": self.check_binlog_config(), "gtid_config": self.check_gtid_config(), "innodb_config": self.check_innodb_config(), "performance": self.check_performance(), } return health_check_result except Error as e: self.logger.error(f"运行健康检查时出错: {e}") return {"error": str(e)} finally: self.disconnect()