from datetime import datetime import sys import logging import os import pymysql # 配置日志格式和文件名 LOG_DIR = '/home/oracle/log' log_filename = f'auto_statistics_collection_{datetime.now().strftime("%Y%m%d")}.log' log_file = os.path.join(LOG_DIR, log_filename) # 确保日志目录存在 os.makedirs(LOG_DIR, exist_ok=True) # 配置日志，设置为追加模式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, mode='a') ] ) logger = logging.getLogger(__name__) # 获取今天的日期 current_date = datetime.today().date() current_date_str = current_date.strftime('%Y%m%d') def ob_mysql_partition_statistics_collect(cursor_ob_db): try: logger.info(f"{current_date_str}") # 查询当月分区POSITION query_partition = """select table_name,min(PARTITION_POSITION) from (select * from (select table_owner,table_name,PARTITION_NAME,PARTITION_POSITION, replace(substring(replace(HIGH_VALUE,"'",''),1,10),'-','') as time_format from dba_tab_partitions where table_owner='table_owner') a where a.time_format >= %s) group by table_owner,table_name""" logger.info("执行查询分区信息SQL") cursor_ob_db.execute(query_partition,(current_date_str,)) partitions = cursor_ob_db.fetchall() logger.info(f"查询当月分区信息完成") if partitions is None or not partitions: logger.info(f"不存在分区表") return None else: logger.info(f"待处理 {len(partitions)} 个分区") for partition in partitions: # 查询当月分区号 query_partition_name = """select PARTITION_NAME from dba_tab_partitions where table_owner='table_owner' and table_name=%s and PARTITION_POSITION=%s""" logger.info(f"查询参数：{partition[0]},{partition[1]}") cursor_ob_db.execute(query_partition_name, (partition[0], partition[1])) partition_name_result = cursor_ob_db.fetchone() if not partition_name_result: logger.error(f"未能获取到分区名，跳过该分区") continue logger.info(f" 表: {partition[0]} , 当月分区名: {partition_name_result[0]}") table_name_real = partition[0].strip() partition_name_real = partition_name_result[0].strip() if partition[0] and partition_name_result[0] and table_name_real == partition[0] and partition_name_real == partition_name_result[0]: # 查询最近收集时间 check_before_query = """select max(replace(replace(replace(substring(LAST_ANALYZED,1,19),'-',''),':',''),' ','')) from dba_part_col_statistics where OWNER='table_owner' and TABLE_NAME=%s and PARTITION_NAME=%s""" cursor_ob_db.execute(check_before_query,(partition[0], partition_name_result[0])) check_before_result = cursor_ob_db.fetchone() # 处理查询结果 if check_before_result and check_before_result[0]: check_before_result_value = check_before_result[0] else: check_before_result_value = None logger.info(f"最近收集时间查询完成为: {check_before_result_value}") # 统计信息收集 auto_statistics_collect_sql = """ CALL dbms_stats.gather_table_stats(ownname=>'table_owner', tabname=>%s, partname=>%s,granularity=>'APPROX_GLOBAL AND PARTITION') """ logger.info(f"统计信息收集带入参数：{partition[0]},{partition_name_result[0]}") try: cursor_ob_db.execute(auto_statistics_collect_sql, (partition[0], partition_name_result[0])) logger.info(f"统计信息收集执行完成") except Exception as e: logger.error(f"统计信息收集执行失败: {e}") continue # 查询最新收集时间 check_after_query = """select max(replace(replace(replace(substring(LAST_ANALYZED,1,19),'-',''),':',''),' ','')) from dba_part_col_statistics where OWNER='table_owner' and TABLE_NAME=%s and PARTITION_NAME=%s""" cursor_ob_db.execute(check_after_query,(partition[0], partition_name_result[0])) check_after_result = cursor_ob_db.fetchone() # 处理查询结果 if check_after_result and check_after_result[0]: check_after_result_value = check_after_result[0] else: check_after_result_value = None logger.info(f"最新收集时间为:{check_after_result_value}") # 比较时间 if check_after_result_value and check_before_result_value: if check_after_result_value > check_before_result_value: logger.info(f"表{partition[0]} 统计信息收集成功，收集时间从 {check_before_result_value} 更新为 {check_after_result_value}") else: logger.error(f"表{partition[0]} 统计信息收集失败，当前收集时间为{check_before_result_value}") elif check_before_result_value is None: logger.info(f"表{partition[0]} 之前没有统计信息，现在已收集") else: logger.error(f"表{partition[0]} 统计信息收集失败，无法获取新时间") else: logger.error(f"表信息存在问题，跳过该表: {partition[0]}") return partitions except Exception as e: logger.error(f"查询分区信息出错:{e}") return None if __name__ == '__main__': try: logger.info(f"OB-Mysql租户table_owner数据库自动统计信息收集开始:") try: username = f'insp_user@租户名#obarchprd' db_config = { 'host': 'ip', 'port': 2883, 'user': username, 'password': 'password', 'database': 'oceanbase' } conn_ob_mysql = pymysql.connect(**db_config) cursor_ob_db = conn_ob_mysql.cursor() partitions = ob_mysql_partition_statistics_collect(cursor_ob_db) cursor_ob_db.close() conn_ob_mysql.close() except Exception as e: logger.error(f"MySQL连接或查询失败: {e}") except ValueError as e: logger.critical(f"错误: {e}") except Exception as e: logger.critical(f"未知错误: {e}")再编写一个shell脚本调用python脚本，配置crontab即可开启自动统计信息收集
此脚本支持MySql8 和 MySql5 版本在“记录结果的ip”的mysql中建表，所有巡检结果都放在该表中，后续可以配合zabbix做定期巡检create table binlog_bigtrx_records ( id int not null auto_increment, mysql_ip varchar(30), min_binlog varchar(30), max_binlog varchar(30), record_time timestamp null default current_timestamp, update_time timestamp null default current_timestamp on update current_timestamp, exist_big_trx varchar(10), primary key (id));binlog_bigtrx_show.py''' 要求： 1、将事务大小存于表中，并且记录上一次记录的最大binlog号，下一次检查从该binlog开始 2、并且增加巡检脚本 表内容： ip、开始binlog、结束binlog、库名、表名、事务大小、操作类型、操作事件 1、根据最小最大binlog号，得出所有binlog号（根据update_time判断是否当天更新，未更新的不检查事务） 2、遍历所有binlog号，将大事务记录到表中（每天更新），生成日志（每天汇总一条） 3、 ''' import argparse import sys from datetime import datetime import pymysql import logging import os from datetime import datetime # 配置日志格式和文件名 LOG_DIR = '/home/mysql/mysql_big_trx_check/log' log_filename = f'binlog_event_monitor_{datetime.now().strftime("%Y%m%d")}.log' log_file = os.path.join(LOG_DIR, log_filename) # 确保日志目录存在 os.makedirs(LOG_DIR, exist_ok=True) # 配置日志，设置为追加模式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, mode='a'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) curr_dt = datetime.now().strftime("%Y%m%d_%H%M%S") db_source_config = { 'host': '记录结果的ip', 'port': 3306, 'user': 'binlogxunjian', 'password': 'binlogxunjian', 'database': 'blxjdb' } def append_file(file_name, content): # 以写入的方式打开 f = open(file_name, 'a+') # 写入内容 f.write(content) # 换行符 f.write('\n') # 关闭文件 f.close() def get_event_from_binlog_by_sql(binlog, mysql_ip, mysql_port, user_name, password, trx_min_size): logger.info(f"{mysql_ip}:正在解析日志: {binlog}") result_file = f"/home/mysql/mysql_big_trx_check/big_trx_info/event_result_{mysql_ip}_{binlog}_{curr_dt}.log" dbconn = pymysql.connect(host=mysql_ip, user=user_name, passwd=password, port=mysql_port) # use_unicode=False, cur = dbconn.cursor() show_binlog_events_sql = f"show binlog events in '{binlog}'" cur.execute(show_binlog_events_sql) binlog_events = cur.fetchall() if len(binlog_events) > 0: last_event_type = '' last_tab_name = '' last_event_size = 0 # 大事务发现标志 bigtrx_found = False for bv_row in binlog_events: try: log_name, start_pos, event_type, server_id, end_pos, info = bv_row if info[0:5].upper() == 'BEGIN': logger.debug(f"Transaction started at position: {start_pos}") # tab_fname_list = [] trx_start_pos = start_pos update_events_size = {} # update_events_count = {} delete_events_size = {} # delete_events_count = {} write_events_size = {} # write_events_count = {} trx_size = 0 logger.debug(f"Transaction started at position: {trx_start_pos}") if event_type.upper() == 'TABLE_MAP': tab_fname = info.split()[2].replace('(', '').replace(')', '') logger.debug(f"Mapped table: {tab_fname}") if event_type.upper() == 'UPDATE_ROWS': trx_size += end_pos - start_pos if tab_fname in update_events_size: update_events_size[tab_fname] = update_events_size[tab_fname] + end_pos - start_pos # update_events_count[tab_name] = update_events_count[tab_name] + 1 else: update_events_size[tab_fname] = end_pos - start_pos # update_events_count[tab_name] = 1 if event_type.upper() == 'DELETE_ROWS': trx_size += end_pos - start_pos if tab_fname in delete_events_size: delete_events_size[tab_fname] = delete_events_size[tab_fname] + end_pos - start_pos # delete_events_count[tab_name] = delete_events_count[tab_name] + 1 else: delete_events_size[tab_fname] = end_pos - start_pos # delete_events_count[tab_name] = 1 if event_type.upper() == 'WRITE_ROWS': trx_size += end_pos - start_pos if tab_fname in write_events_size: write_events_size[tab_fname] = write_events_size[tab_fname] + end_pos - start_pos # write_events_count[tab_name] = write_events_count[tab_name] + 1 else: write_events_size[tab_fname] = end_pos - start_pos # write_events_count[tab_name] = 1 if info[0:6].upper() == 'COMMIT': logger.debug(f"Transaction committed at position: {end_pos}") trx_end_pos = end_pos if trx_size / 1024 / 1024 > trx_min_size: bigtrx_found = True output_str = f"{mysql_ip};{round((trx_end_pos - trx_start_pos) / 1024 / 1024, 2)} MB;" if update_events_size: for tab_fname1 in update_events_size: db_name = tab_fname1.split('.')[0].strip().strip('`') tab_bname = tab_fname1.split('.')[1].strip().strip('`') output_str += f"UPD-size:{update_events_size};trx_start_pos:{trx_start_pos};'### UPDATE `{db_name}`.`{tab_bname}`';" if delete_events_size: for tab_fname1 in delete_events_size: db_name = tab_fname1.split('.')[0].strip().strip('`') tab_bname = tab_fname1.split('.')[1].strip().strip('`') output_str += f"DLT-size:{delete_events_size};trx_start_pos:{trx_start_pos};'### DELETE FROM `{db_name}`.`{tab_bname}`';" if write_events_size: for tab_fname1 in write_events_size: db_name = tab_fname1.split('.')[0].strip().strip('`') tab_bname = tab_fname1.split('.')[1].strip().strip('`') output_str += f"WTE-size:{write_events_size};trx_start_pos:{trx_start_pos};'### INSERT INTO `{db_name}`.`{tab_bname}`';" append_file(result_file, output_str) except Exception as e: logger.error(f"{mysql_ip}:Parse event:{bv_row} is error, Detail is :{e}.") sys.exit(1) if bigtrx_found: logger.info(f"{mysql_ip}:Full event result:[ {result_file} ]") else: logger.error(f"{mysql_ip}:binlog file:{binlog} is invalid") # print(f"Full event result:[ {result_file} ]") dbconn.close() if __name__ == '__main__': parser = argparse.ArgumentParser(description='A piece of binlog analysis code.') parser.add_argument('--binlog', type=str, help="binlog file(Run as [show binlog events in 'mysql-bin.00001' ).", default=None) parser.add_argument('--mysql_ip', type=str, help='mysql database ip addr.', default=None) parser.add_argument('--port', type=int, help='mysql database ip addr.', default=3306) parser.add_argument('--uname', type=str, help='mysql database username.', default='xjuser') parser.add_argument('--password', type=str, help='mysql database password.', default = None) parser.add_argument('--trx_min_size', type=int, help='need print min trx binlog size(MB).', default=1) args = parser.parse_args() if not args.binlog or not args.mysql_ip: logger.error("Missing required arguments: --binlog and --mysql_ip") parser.print_help() sys.exit(0) try: password = 'password' get_event_from_binlog_by_sql(args.binlog, args.mysql_ip, args.port, args.uname, password, args.trx_min_size) logger.info(f"{args.mysql_ip}:{args.binlog}解析完成！") except Exception as e: logger.error(f"{args.mysql_ip}:解析过程中出现错误: {e}") sys.exit(1)binlog_list.pyimport argparse import sys from datetime import datetime import pymysql import logging import os from datetime import datetime # 配置日志格式和文件名 LOG_DIR = '/home/mysql/mysql_big_trx_check/log' log_filename = f'binlog_event_monitor_{datetime.now().strftime("%Y%m%d")}.log' log_file = os.path.join(LOG_DIR, log_filename) # 确保日志目录存在 os.makedirs(LOG_DIR, exist_ok=True) # 配置日志，设置为追加模式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, mode='a'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) curr_dt = datetime.now().strftime("%Y-%m-%d") def binlog_min_max_query(cursor_source, mysql_ip, curr_dt): try: sql = "select min_binlog,max_binlog from blxjdb.binlog_bigtrx_records where mysql_ip = %s and DATE(record_time) = %s order by record_time desc limit 1" cursor_source.execute(sql,(mysql_ip, curr_dt)) result = cursor_source.fetchone() if result: min_binlog, max_binlog = result[0], result[1] return min_binlog, max_binlog else: logger.info(f"{mysql_ip}:没有找到对应日期的binlog记录") return None, None except Exception as e: logger.error(f"{mysql_ip}:查询binlog_min_max时出错: {e}") return None, None def get_binlog_list(mysql_ip, min_binlog, max_binlog): index = len(min_binlog) while index > 0 and min_binlog[index - 1].isdigit(): index -= 1 min_binlog_num = int(min_binlog[index:]) max_binlog_num = int(max_binlog[index:]) binlog_list = [] for i in range(min_binlog_num, max_binlog_num + 1): binlog = f"mysql-bin.{i:06d}" binlog_list.append(binlog) print(binlog) logger.info(f"{mysql_ip}:待解析binlog日志列表：{binlog_list}") return binlog_list def main(mysql_ip): db_source_config = { 'host': '记录结果的ip', 'port': 3306, 'user': 'binlogxunjian', 'password': 'binlogxunjian', 'database': 'blxjdb' } try: conn_source = pymysql.connect(**db_source_config) cursor_source = conn_source.cursor() min_binlog, max_binlog = binlog_min_max_query(cursor_source, mysql_ip, curr_dt) if min_binlog is None or max_binlog is None: logger.warning(f"{mysql_ip}:未能获取有效的binlog范围。") return binlog_list = get_binlog_list(mysql_ip, min_binlog, max_binlog) # for binlog in binlog_list: # print(binlog) except Exception as e: logger.error(f"{mysql_ip}:主函数执行时出错: {e}") finally: # 确保关闭数据库连接和游标 if cursor_source: cursor_source.close() if conn_source: conn_source.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description='Get binlog list.') parser.add_argument('--mysql_ip', type=str, help='mysql database ip addr.', default=None) args = parser.parse_args() if not args.mysql_ip: parser.print_help() sys.exit(0) main(args.mysql_ip)binlog_records.pyimport argparse import sys import pymysql import logging import os import re from datetime import datetime # 配置日志格式和文件名 LOG_DIR = '/home/mysql/mysql_big_trx_check/log' log_filename = f'binlog_event_monitor_{datetime.now().strftime("%Y%m%d")}.log' log_file = os.path.join(LOG_DIR, log_filename) # 确保日志目录存在 os.makedirs(LOG_DIR, exist_ok=True) # 配置日志，设置为追加模式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, mode='a'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) def get_binlog_min_max(cursor_target, mysql_ip): """获取指定 binlog 文件的最小和最大位置""" try: query_binlog = f"show binary logs;" #query_binlog = f"select * from binary_logs;" cursor_target.execute(query_binlog) binlog = cursor_target.fetchall() if binlog is None or not binlog: logger.error(f"{mysql_ip}:Binlog 文件不存在或无法读取。") return None # 自定义排序函数 def binlog_sort_key(binlog_name): match = re.search(r'(\d+)$', binlog_name) if match: return int(match.group(1)) return 0 # 对 binlog 按文件名排序 sort_binlog = sorted(binlog, key=lambda x: binlog_sort_key(x[0])) print(sort_binlog) # 解析事件，找到最小和最大位置 min_binlog = sort_binlog[0][0] max_binlog = sort_binlog[-1][0] return min_binlog, max_binlog except Exception as e: logger.error(f"{mysql_ip}:获取 binlog 最小和最大值时出错: {e}") return None def insert_binlog_record(cursor_source, mysql_ip, min_binlog, max_binlog): """将 binlog 的最小和最大位置插入到记录表中""" try: insert = """ INSERT INTO blxjdb.binlog_bigtrx_records (mysql_ip, min_binlog, max_binlog) VALUES (%s, %s, %s) """ cursor_source.execute(insert, (mysql_ip, min_binlog, max_binlog)) except Exception as e: logger.error(f"{mysql_ip}:插入记录时出错: {e}") def get_last_max_binlog(cursor_source, mysql_ip): """获取最后一次记录的最大位置""" try: query = """ SELECT max(max_binlog) FROM blxjdb.binlog_bigtrx_records WHERE mysql_ip = %s """ cursor_source.execute(query, (mysql_ip,)) result = cursor_source.fetchall() if result is None or len(result) == 0: logger.info(f"{mysql_ip}:没有找到上次的 binlog 最大值记录") return None max_binlog = result[0][0] logger.info(f"{mysql_ip}:上次记录的最大 binlog 位置为: {max_binlog}") return max_binlog except Exception as e: logger.error(f"{mysql_ip}:获取上次最大值时出错: {e}") return None def main(mysql_ip): db_source_config = { 'host': '记录结果的ip', 'port': 3306, 'user': 'binlogxunjian', 'password': 'binlogxunjian' } db_target_config = { 'host': mysql_ip, 'port': 3306, 'user': 'username', 'password': 'password' } #db_target_config['host'] = mysql_ip conn_source = pymysql.connect(**db_source_config) cursor_source = conn_source.cursor() conn_target = pymysql.connect(**db_target_config) cursor_target = conn_target.cursor() try: # 获取最后一次记录的最大位置 last_max = get_last_max_binlog(cursor_source, mysql_ip) # 获取当前 binlog 文件的最小和最大位置 min_binlog, max_binlog = get_binlog_min_max(cursor_target, mysql_ip) if min_binlog is None or max_binlog is None: return # 确定新的最小位置 if last_max is None: # 无记录则插入 logger.info(f"{mysql_ip}:没有记录，开始新数据插入！") insert_binlog_record(cursor_source, mysql_ip, min_binlog, max_binlog) logger.info(f"{mysql_ip}:新数据插入完成！") conn_source.commit() elif last_max < max_binlog or last_max == max_binlog: new_min = last_max # last_max < max_binlog 则更新 insert_binlog_record(cursor_source, mysql_ip, new_min, max_binlog) conn_source.commit() logger.info(f"{mysql_ip}:binlog存在变化!") else: logger.info(f"{mysql_ip}:没有新的 binlog 变化！") except Exception as e: logger.error(f"{mysql_ip}:主函数执行时出错: {e}") conn_source.rollback() finally: if conn_source: conn_source.close() cursor_source.close() if conn_target: conn_target.close() cursor_target.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description='Insert Binlog Records') parser.add_argument('--mysql_ip', type=str, help='mysql database ip addr.', default=None) args = parser.parse_args() if not args.mysql_ip: parser.print_help() sys.exit(0) main(args.mysql_ip)binlog_event_analyze.sh#!/bin/bash workdir="/home/oracle/monitor/mysql_big_trx_check" cd $workdir #调用py3环境 source /home/oracle/py3/bin/activate log_bigtrx_dir=/home/oracle/monitor/mysql_big_trx_check/big_trx_info curr_dt=$(date +%Y%m%d) export MYSQL_PWD=binlogxunjian #设置最小事务大小 trx_min_size=90 mysql_ip=$1 function update_binlog_bigtrx_records() { local mysql_ip=$1 local exist_big_trx=$2 mysql -ubinlogxunjian -hip --silent --skip-column-names -e \ "update blxjdb.binlog_bigtrx_records set exist_big_trx='$exist_big_trx' where mysql_ip='$mysql_ip' and date_format(record_time, '%Y%m%d')=${curr_dt};" } # 调用Python脚本获取binlog记录 python binlog_records.py --mysql_ip $mysql_ip binlog_list=$(python binlog_list.py --mysql_ip $mysql_ip | tr '\n' ' ') # 记录是否存在大事务 for binlog in $binlog_list;do # 将任务添加到作业队列 # { python binlog_bigtrx_show.py --mysql_ip $mysql_ip --binlog $binlog --trx_min_size $trx_min_size # } & done if [ "$(ls ${log_bigtrx_dir}/event_result_${mysql_ip}_*_${curr_dt}_*.log 2> /dev/null |wc -l)" -gt 0 ];then update_binlog_bigtrx_records $mysql_ip 'Y' else update_binlog_bigtrx_records $mysql_ip 'N' fibinlog_event_monitor.shMysql8版本的待巡检Mysql的ip是记录在oracle库的表中的，可根据实际要求修改下面的Mysql5版本的待巡检Mysql的ip是写死在文件中的，可进行参考#!/bin/bash # 日志记录 curr_dt=`date +%Y%m%d` log_filename="/home/mysql/mysql_big_trx_check/log/binlog_event_monitor_${curr_dt}.log" # 最大并行任务数 MAX_PARALLEL=4 # 任务超时时间（秒） TIMEOUT=7200 # 例如，设置为120分钟 # Oracle 数据库连接信息 ORACLE_USER="user" ORACLE_PASS="password" ORACLE_SID="orcl" # 从Oracle表中获取MySQL IP列表 mysql_ips=$(sqlplus -S ${ORACLE_USER}/${ORACLE_PASS}@ip:1521/${ORACLE_SID} <<EOF set heading off set feedback off set pagesize 0 set timing off select HOST from MYSQL_HOST; exit; EOF ) # 存储待处理的任务队列 queue=($mysql_ips) # 存储正在运行的任务PID及其启动时间 declare -A tasks=() # 使用关联数组，键为PID，值为启动时间 # 陷阱处理，捕获终止信号 trap ctrl_c INT TERM # 终止信号处理函数 function ctrl_c() { # 终止所有子进程 for pid in "${!tasks[@]}"; do if kill -0 "$pid" 2>/dev/null; then kill "$pid" unset tasks["$pid"] echo "ERROR: 终止任务PID: $pid" >> $log_filename fi done echo "ERROR: 脚本终止，所有子进程已关闭。" >> $log_filename exit 1 } # 遍历每个MySQL IP进行解析 function binlog_event_analyze() { local mysql_ip=$1 echo "现在开始解析: $mysql_ip" >> $log_filename binlog_event_analyze.sh "$mysql_ip" & new_pid=$! curr_time=`date +%Y%m%d_%H%M%S` current_time=$(date +%s) tasks["$new_pid"]=$current_time echo "$mysql_ip: 任务PID: $new_pid, 启动时间: $curr_time" >> $log_filename } # 检查任务状态并处理超时 function check_task_status() { local current_time=$(date +%s) for pid in "${!tasks[@]}"; do if ! kill -0 "$pid" 2>/dev/null; then unset tasks["$pid"] echo "$mysql_ip: 任务PID: $pid 已完成或不存在，已释放资源。" >> $log_filename else # 检查任务是否超时 start_time=${tasks[$pid]} if [ $((current_time - start_time)) -ge $TIMEOUT ]; then kill "$pid" unset tasks["$pid"] echo "ERROR: $mysql_ip: 任务PID: $pid 超时终止，已释放资源。" >> $log_filename fi fi done } # 主循环：遍历所有MySQL IP log_interval=120 log_count=0 # 主循环：遍历所有MySQL IP while [ ${#queue[@]} -gt 0 ]; do # 检查任务状态并处理超时 check_task_status # 启动新任务 if [ "${#tasks[@]}" -lt "$MAX_PARALLEL" ]; then mysql_ip=${queue[0]} binlog_event_analyze "$mysql_ip" queue=("${queue[@]:1}") # 移除已启动的任务 log_count=0 else # 每120次(1min)循环输出一次队列状态 log_count=$((log_count + 1)) if [ $log_count -ge $log_interval ]; then echo "并行任务已达最大值$MAX_PARALLEL，请稍等..." >> $log_filename echo "待处理的IP队列: ${queue[@]}" >> $log_filename log_count=0 fi fi # 短暂等待，避免过于频繁的进程检查 sleep 0.5 done # 等待所有任务完成 for pid in "${!tasks[@]}"; do wait "$pid" 2>/dev/null unset tasks["$pid"] done echo "所有任务已完成。" >> $log_filename以下是针对MySql5版本的脚本（binlog_list.py 复用MySql8版本的）binlog_bigtrx_show_for_mysql5.pyimport argparse import sys from datetime import datetime import pymysql import logging import os # 配置日志格式和文件名 LOG_DIR = '/home/mysql/mysql_big_trx_check/log' log_filename = f'binlog_event_monitor_{datetime.now().strftime("%Y%m%d")}.log' log_file = os.path.join(LOG_DIR, log_filename) # 确保日志目录存在 os.makedirs(LOG_DIR, exist_ok=True) # 配置日志，设置为追加模式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, mode='a'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) curr_dt = datetime.now().strftime("%Y%m%d_%H%M%S") db_source_config = { 'host': '记录结果的ip', 'port': 3306, 'user': 'binlogxunjian', 'password': 'binlogxunjian', 'database': 'blxjdb' } def append_file(file_name, content): # 以写入的方式打开 f = open(file_name, 'a+') # 写入内容 f.write(content) # 换行符 f.write('\n') # 关闭文件 f.close() def get_event_from_binlog_by_sql(binlog, mysql_ip, mysql_port, user_name, password, trx_min_size): logger.info(f"{mysql_ip}:正在解析日志: {binlog}") result_file = f"/home/mysql/mysql_big_trx_check/big_trx_info/event_result_{mysql_ip}_{binlog}_{curr_dt}.log" dbconn = pymysql.connect(host=mysql_ip, user=user_name, passwd=password, port=mysql_port) # use_unicode=False, cur = dbconn.cursor() show_binlog_events_sql = f"show binlog events in '{binlog}'" cur.execute(show_binlog_events_sql) binlog_events = cur.fetchall() if len(binlog_events) > 0: last_event_type = '' last_tab_name = '' last_event_size = 0 # 大事务发现标志 bigtrx_found = False for bv_row in binlog_events: try: log_name, start_pos, event_type, server_id, end_pos, info = bv_row if info[0:5].upper() == 'BEGIN': logger.debug(f"Transaction started at position: {start_pos}") # tab_fname_list = [] trx_start_pos = start_pos update_events_size = {} # update_events_count = {} delete_events_size = {} # delete_events_count = {} write_events_size = {} # write_events_count = {} trx_size = 0 logger.debug(f"Transaction started at position: {trx_start_pos}") if event_type.upper() == 'TABLE_MAP': tab_fname = info.split()[2].replace('(', '').replace(')', '') logger.debug(f"Mapped table: {tab_fname}") if event_type.upper() == 'UPDATE_ROWS': trx_size += end_pos - start_pos if tab_fname in update_events_size: update_events_size[tab_fname] = update_events_size[tab_fname] + end_pos - start_pos # update_events_count[tab_name] = update_events_count[tab_name] + 1 else: update_events_size[tab_fname] = end_pos - start_pos # update_events_count[tab_name] = 1 if event_type.upper() == 'DELETE_ROWS': trx_size += end_pos - start_pos if tab_fname in delete_events_size: delete_events_size[tab_fname] = delete_events_size[tab_fname] + end_pos - start_pos # delete_events_count[tab_name] = delete_events_count[tab_name] + 1 else: delete_events_size[tab_fname] = end_pos - start_pos # delete_events_count[tab_name] = 1 if event_type.upper() == 'WRITE_ROWS': trx_size += end_pos - start_pos if tab_fname in write_events_size: write_events_size[tab_fname] = write_events_size[tab_fname] + end_pos - start_pos # write_events_count[tab_name] = write_events_count[tab_name] + 1 else: write_events_size[tab_fname] = end_pos - start_pos # write_events_count[tab_name] = 1 if info[0:6].upper() == 'COMMIT': logger.debug(f"Transaction committed at position: {end_pos}") trx_end_pos = end_pos if trx_size / 1024 / 1024 > trx_min_size: bigtrx_found = True output_str = f"{mysql_ip};{round((trx_end_pos - trx_start_pos) / 1024 / 1024, 2)} MB;" if update_events_size: for tab_fname1 in update_events_size: db_name = tab_fname1.split('.')[0].strip().strip('`') tab_bname = tab_fname1.split('.')[1].strip().strip('`') output_str += f"UPD-size:{update_events_size};trx_start_pos:{trx_start_pos};'### UPDATE `{db_name}`.`{tab_bname}`';" if delete_events_size: for tab_fname1 in delete_events_size: db_name = tab_fname1.split('.')[0].strip().strip('`') tab_bname = tab_fname1.split('.')[1].strip().strip('`') output_str += f"DLT-size:{delete_events_size};trx_start_pos:{trx_start_pos};'### DELETE FROM `{db_name}`.`{tab_bname}`';" if write_events_size: for tab_fname1 in write_events_size: db_name = tab_fname1.split('.')[0].strip().strip('`') tab_bname = tab_fname1.split('.')[1].strip().strip('`') output_str += f"WTE-size:{write_events_size};trx_start_pos:{trx_start_pos};'### INSERT INTO `{db_name}`.`{tab_bname}`';" append_file(result_file, output_str) except Exception as e: logger.error(f"{mysql_ip}:Parse event:{bv_row} is error, Detail is :{e}.") sys.exit(1) if bigtrx_found: logger.info(f"{mysql_ip}:Full event result:[ {result_file} ]") else: logger.error(f"{mysql_ip}:binlog file:{binlog} is invalid") # print(f"Full event result:[ {result_file} ]") dbconn.close() if __name__ == '__main__': parser = argparse.ArgumentParser(description='A piece of binlog analysis code.') parser.add_argument('--binlog', type=str, help="binlog file(Run as [show binlog events in 'mysql-bin.00001' ).", default=None) parser.add_argument('--mysql_ip', type=str, help='mysql database ip addr.', default=None) parser.add_argument('--port', type=int, help='mysql database ip addr.', default=3306) parser.add_argument('--uname', type=str, help='mysql database username.', default='blxjuser') parser.add_argument('--password', type=str, help='mysql database password.', default = None) parser.add_argument('--trx_min_size', type=int, help='need print min trx binlog size(MB).', default=1) args = parser.parse_args() if not args.binlog or not args.mysql_ip: logger.error("Missing required arguments: --binlog and --mysql_ip") parser.print_help() sys.exit(0) try: password = 'password' get_event_from_binlog_by_sql(args.binlog, args.mysql_ip, args.port, args.uname, password, args.trx_min_size) logger.info(f"{args.mysql_ip}:{args.binlog}解析完成！") except Exception as e: logger.error(f"{args.mysql_ip}:解析过程中出现错误: {e}") sys.exit(1)binlog_records_for_mysql5.pyimport argparse import sys import pymysql import logging import os import re from datetime import datetime # 配置日志格式和文件名 LOG_DIR = '/home/mysql/mysql_big_trx_check/log' log_filename = f'binlog_event_monitor_{datetime.now().strftime("%Y%m%d")}.log' log_file = os.path.join(LOG_DIR, log_filename) # 确保日志目录存在 os.makedirs(LOG_DIR, exist_ok=True) # 配置日志，设置为追加模式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, mode='a'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) def get_binlog_min_max(cursor_target, mysql_ip): """获取指定 binlog 文件的最小和最大位置""" try: query_binlog = f"show binary logs;" #query_binlog = f"select * from binary_logs;" cursor_target.execute(query_binlog) binlog = cursor_target.fetchall() if binlog is None or not binlog: logger.error(f"{mysql_ip}:Binlog 文件不存在或无法读取。") return None # 自定义排序函数 def binlog_sort_key(binlog_name): match = re.search(r'(\d+)$', binlog_name) if match: return int(match.group(1)) return 0 # 对 binlog 按文件名排序 sort_binlog = sorted(binlog, key=lambda x: binlog_sort_key(x[0])) print(sort_binlog) # 解析事件，找到最小和最大位置 min_binlog = sort_binlog[0][0] max_binlog = sort_binlog[-1][0] return min_binlog, max_binlog except Exception as e: logger.error(f"{mysql_ip}:获取 binlog 最小和最大值时出错: {e}") return None def insert_binlog_record(cursor_source, mysql_ip, min_binlog, max_binlog): """将 binlog 的最小和最大位置插入到记录表中""" try: insert = """ INSERT INTO blxjdb.binlog_bigtrx_records (mysql_ip, min_binlog, max_binlog) VALUES (%s, %s, %s) """ cursor_source.execute(insert, (mysql_ip, min_binlog, max_binlog)) except Exception as e: logger.error(f"{mysql_ip}:插入记录时出错: {e}") def get_last_max_binlog(cursor_source, mysql_ip): """获取最后一次记录的最大位置""" try: query = """ SELECT max_binlog FROM blxjdb.binlog_bigtrx_records WHERE mysql_ip = %s """ cursor_source.execute(query, (mysql_ip,)) result = cursor_source.fetchall() if result is None or len(result) == 0: logger.info(f"{mysql_ip}:没有找到上次的 binlog 最大值记录") return None max_binlog = result[0][0] logger.info(f"{mysql_ip}:上次记录的最大 binlog 位置为: {max_binlog}") return max_binlog except Exception as e: logger.error(f"{mysql_ip}:获取上次最大值时出错: {e}") return None def main(mysql_ip): db_source_config = { 'host': '记录结果的ip', 'port': 3306, 'user': 'binlogxunjian', 'password': 'binlogxunjian' } db_target_config = { 'host': mysql_ip, 'port': 3306, 'user': 'blxjuser', 'password': 'password' } #db_target_config['host'] = mysql_ip conn_source = pymysql.connect(**db_source_config) cursor_source = conn_source.cursor() conn_target = pymysql.connect(**db_target_config) cursor_target = conn_target.cursor() try: # 获取最后一次记录的最大位置 last_max = get_last_max_binlog(cursor_source, mysql_ip) # 获取当前 binlog 文件的最小和最大位置 min_binlog, max_binlog = get_binlog_min_max(cursor_target, mysql_ip) if min_binlog is None or max_binlog is None: return # 确定新的最小位置 if last_max is None: # 无记录则插入 logger.info(f"{mysql_ip}:没有记录，开始新数据插入！") insert_binlog_record(cursor_source, mysql_ip, min_binlog, max_binlog) logger.info(f"{mysql_ip}:新数据插入完成！") conn_source.commit() elif last_max < max_binlog or last_max == max_binlog: new_min = last_max # last_max < max_binlog 则更新 insert_binlog_record(cursor_source, mysql_ip, new_min, max_binlog) conn_source.commit() logger.info(f"{mysql_ip}:binlog存在变化!") else: logger.info(f"{mysql_ip}:没有新的 binlog 变化！") except Exception as e: logger.error(f"{mysql_ip}:主函数执行时出错: {e}") conn_source.rollback() finally: if conn_source: conn_source.close() cursor_source.close() if conn_target: conn_target.close() cursor_target.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description='Insert Binlog Records') parser.add_argument('--mysql_ip', type=str, help='mysql database ip addr.', default=None) args = parser.parse_args() if not args.mysql_ip: parser.print_help() sys.exit(0) main(args.mysql_ip)binlog_event_analyze_for_mysql5.sh#!/bin/bash workdir="/home/mysql/mysql_big_trx_check" cd $workdir #调用py3环境 source /home/oracle/py3/bin/activate log_bigtrx_dir=/home/mysql/mysql_big_trx_check/big_trx_info curr_dt=$(date +%Y%m%d) export MYSQL_PWD=binlogxunjian #设置最小事务大小 trx_min_size=90 mysql_ip=$1 function update_binlog_bigtrx_records() { local mysql_ip=$1 local exist_big_trx=$2 mysql -ubinlogxunjian -h记录结果的ip --silent --skip-column-names -e \ "update blxjdb.binlog_bigtrx_records set exist_big_trx='$exist_big_trx' where mysql_ip='$mysql_ip' and date_format(record_time, '%Y%m%d')=${curr_dt};" } # 调用Python脚本获取binlog记录 python binlog_records_for_mysql5.py --mysql_ip $mysql_ip binlog_list=$(python binlog_list.py --mysql_ip $mysql_ip | tr '\n' ' ') # 记录是否存在大事务 for binlog in $binlog_list;do # 将任务添加到作业队列 # { python binlog_bigtrx_show_for_mysql5.py --mysql_ip $mysql_ip --binlog $binlog --trx_min_size $trx_min_size # } & done if [ "$(ls ${log_bigtrx_dir}/event_result_${mysql_ip}_*_${curr_dt}_*.log 2> /dev/null |wc -l)" -gt 0 ];then update_binlog_bigtrx_records $mysql_ip 'Y' else update_binlog_bigtrx_records $mysql_ip 'N' fibinlog_event_monitor_for_mysql5.shMysql5版本的待巡检Mysql的ip是写死在文本里的#!/bin/bash # 日志记录 curr_dt=`date +%Y%m%d` log_filename="/home/mysql/mysql_big_trx_check/log/binlog_event_monitor_for_mysql5_${curr_dt}.log" # 最大并行任务数 MAX_PARALLEL=4 # 任务超时时间（秒） TIMEOUT=7200 # 例如，设置为120分钟 # 读取MySQL IP列表 mysql_ips=$(cat /home/mysql/mysql_big_trx_check/mysql_ip.list) # 存储待处理的任务队列 queue=($mysql_ips) # 存储正在运行的任务PID及其启动时间 declare -A tasks=() # 使用关联数组，键为PID，值为启动时间 # 陷阱处理，捕获终止信号 trap ctrl_c INT TERM # 终止信号处理函数 function ctrl_c() { # 终止所有子进程 for pid in "${!tasks[@]}"; do if kill -0 "$pid" 2>/dev/null; then kill "$pid" unset tasks["$pid"] echo "ERROR: 终止任务PID: $pid" >> $log_filename fi done echo "ERROR: 脚本终止，所有子进程已关闭。" >> $log_filename exit 1 } # 遍历每个MySQL IP进行解析 function binlog_event_analyze_for_mysql5() { local mysql_ip=$1 echo "现在开始解析: $mysql_ip" >> $log_filename binlog_event_analyze_for_mysql5.sh "$mysql_ip" & new_pid=$! curr_time=`date +%Y%m%d_%H%M%S` current_time=$(date +%s) tasks["$new_pid"]=$current_time echo "$mysql_ip: 任务PID: $new_pid, 启动时间: $curr_time" >> $log_filename } # 检查任务状态并处理超时 function check_task_status() { local current_time=$(date +%s) for pid in "${!tasks[@]}"; do if ! kill -0 "$pid" 2>/dev/null; then unset tasks["$pid"] echo "$mysql_ip: 任务PID: $pid 已完成或不存在，已释放资源。" >> $log_filename else # 检查任务是否超时 start_time=${tasks[$pid]} if [ $((current_time - start_time)) -ge $TIMEOUT ]; then kill "$pid" unset tasks["$pid"] echo "ERROR: $mysql_ip: 任务PID: $pid 超时终止，已释放资源。" >> $log_filename fi fi done } # 主循环：遍历所有MySQL IP log_interval=120 log_count=0 # 主循环：遍历所有MySQL IP while [ ${#queue[@]} -gt 0 ]; do # 检查任务状态并处理超时 check_task_status # 启动新任务 if [ "${#tasks[@]}" -lt "$MAX_PARALLEL" ]; then mysql_ip=${queue[0]} binlog_event_analyze_for_mysql5 "$mysql_ip" queue=("${queue[@]:1}") # 移除已启动的任务 log_count=0 else # 每120次(1min)循环输出一次队列状态 log_count=$((log_count + 1)) if [ $log_count -ge $log_interval ]; then echo "并行任务已达最大值$MAX_PARALLEL，请稍等..." >> $log_filename echo "待处理的IP队列: ${queue[@]}" >> $log_filename log_count=0 fi fi # 短暂等待，避免过于频繁的进程检查 sleep 0.5 done # 等待所有任务完成 for pid in "${!tasks[@]}"; do wait "$pid" 2>/dev/null unset tasks["$pid"] done echo "所有任务已完成。" >> $log_filename
{callout color="#f0ad4e"}此脚本主要用来对Mysql、Oracle、OceanBase的分区表的分区进行巡检，用于对已过期以及在一个月内即将过期的分区表进行检查，可以配合zabbix等发出告警{/callout}partition_monitor.pyfrom datetime import datetime from ConnDB import ConnOracle import pymysql import re import argparse import sys import logging import os import subprocess # 过滤表 (/home/oracle/skip_tables.list) 录入格式 # oracle -- ip:schema:tablename:oracle # mysql -- ip:schema:tablename:mysql # oceanbase -- tenantname:schema:tablename:oceanbase # 例如： # 192.168.1.10:t1db:table1:mysql # 192.168.1.10:t1db:table2:mysql # 192.168.1.11:t2db:table3:oracle # tenantname:t3db:table4:oceanbase # 配置日志格式和文件名 LOG_DIR = '/home/oracle/log' log_filename = f'partition_monitor_{datetime.now().strftime("%Y%m%d")}.log' log_file = os.path.join(LOG_DIR, log_filename) # 确保日志目录存在 os.makedirs(LOG_DIR, exist_ok=True) # 配置日志，设置为追加模式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, mode='a') ] ) logger = logging.getLogger(__name__) ora_tns_file_path = '/home/oracle/.tnsnames.ora' # 获取今天的日期 current_date = datetime.today().date() current_date_str = current_date.strftime('%Y%m%d') current_date_int = int(current_date_str) # Oracle def parse_tnsnames_ora(ora_tns_file_path, service_name): with open(ora_tns_file_path, 'r') as file: content = file.read() # 使用正则表达式匹配服务名 # adg_orcl1 = (DESCRIPTION =(ADDRESS = (PROTOCOL = TCP)(HOST = 192.168.1.111)(PORT = 1521))(CONNECT_DATA =(SERVER = DEDICATED)(SERVICE_NAME = ORCL))) pattern = re.compile(rf'^{service_name}\s*=\s*\(\s*DESCRIPTION\s*=\s*\(\s*ADDRESS\s*=\s*\(\s*PROTOCOL\s*=\s*TCP\s*\)\s*\(\s*HOST\s*=\s*([^\)]+)\s*\)\s*\(\s*PORT\s*=\s*(\d+)\s*\)\s*\)\s*\(\s*CONNECT_DATA\s*=\s*\(\s*SERVER\s*=\s*DEDICATED\s*\)\s*\(\s*SERVICE_NAME\s*=\s*([^\)]+)\s*\)\s*\)\s*\)', re.MULTILINE | re.DOTALL) match = pattern.search(content) if match: host = match.group(1).strip() port = match.group(2).strip() dbname = match.group(3).strip() return host, port, dbname else: logger.critical(f"{service_name}: 在 {ora_tns_file_path} 中不存在 ") def format_high_value(high_value): date_pattern1 = re.compile(r"TO_DATE\s*\('\s*(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s*', 'SYYYY-MM-DD HH24:MI:SS', 'NLS_CALENDAR=GREGORIAN'\)") match = date_pattern1.search(high_value) if match: date_str = match.group(1) date_obj = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S') return date_obj.strftime('%Y%m%d') else: date_pattern2 = re.compile(r"'(\d{8})'") match = date_pattern2.search(high_value) if match: return match.group(1) else: return high_value def ora_partition_monitor(service_name): try: host, port, dbname = parse_tnsnames_ora(ora_tns_file_path, service_name) #读取跳过表的列表 excluded_tables = set() try: with open('/home/oracle/skip_tables.list', 'r') as f: for line in f: line = line.strip() # 忽略空行和注释的行 if line and not line.startswith('#') and line.endswith('oracle'): parts = line.split(':') if len(parts) >= 3: ip, schema, table = parts[0], parts[1], parts[2] # 只处理当前ip if ip == host: excluded_tables.add((schema, table)) except FileNotFoundError: # 查不到则使用默认配置 pass zytj_db = ConnOracle('query','query',host,port,dbname) # 执行 SQL 查询 status_sql = ''' select 1 from dual ''' ora_sql = ''' select a.table_owner, a.table_name, max(partition_position) from dba_tab_partitions a, dba_part_tables b, dba_tables c where a.table_name = b.table_name and b.table_name = c.table_name and a.table_owner not in ('SYS', 'SYSTEM', 'MDSYS', 'AUDSYS') -- and b.PARTITIONING_TYPE = 'RANGE' and a.high_value is not null and b.INTERVAL is null and a.table_name not in ('PUB_TRAN_LOG_MID') and a.partition_name not like '%MAXVALUE%' group by a.table_owner, a.table_name ''' status_db = zytj_db.select_noheader(status_sql) if status_db is None or len(status_db) == 0: logger.critical(f"{service_name}: 无法获取数据库状态") return else: results = zytj_db.select_noheader(ora_sql) if results is None or len(results) == 0: logger.info(f"{service_name}: 没有分区表或不存在满足条件的分区表") return else: filtered_patitions = [] for result in results: table_owner = result[0] table_name = result[1] # 如果需要跳过，则跳过 if (table_owner, table_name) in excluded_tables: continue filtered_patitions.append(result) for partition in filtered_patitions: table_owner = partition[0] table_name = partition[1] partition_position = partition[2] detail_sql = f''' select table_owner, table_name, partition_position, partition_name, high_value from dba_tab_partitions where table_owner = '{table_owner}' and table_name = '{table_name}' and partition_position = {partition_position} ''' detail_results = zytj_db.select_noheader(detail_sql) for detail_result in detail_results: table_owner = detail_result[0] table_name = detail_result[1] partition_position = detail_result[2] partition_name = detail_result[3] high_value = detail_result[4] formatted_high_value = format_high_value(high_value) # 将格式化后的 high_value 转换为整数以便比较,设置为30内过期就告警 try: formatted_high_value_int = int(formatted_high_value) if formatted_high_value_int < 19000000 or formatted_high_value_int > 30000000: logger.error(f"{host}:库名:{table_owner},表名:{table_name},目前最大分区为:{partition_name},分区描述:{high_value},分区最高范围至:{formatted_high_value},分区异常") elif formatted_high_value_int > 19000000 and formatted_high_value_int < current_date_int: logger.error(f"{host}:库名:{table_owner},表名:{table_name},目前最大分区为:{partition_name},分区描述:{high_value},分区最高范围至:{formatted_high_value},分区已过期") elif formatted_high_value_int - current_date_int <= 30: logger.warning(f"{host}:库名:{table_owner},表名:{table_name},目前最大分区为:{partition_name},分区描述:{high_value},分区最高范围至:{formatted_high_value},需及时建立新分区") else: logger.info(f"{host}:库名:{table_owner},表名:{table_name},目前最大分区为:{partition_name},分区描述:{high_value},分区最高范围至:{formatted_high_value},分区正常") except ValueError: formatted_high_value_int = 0 # 如果无法转换为整数，则设为 0 zytj_db.disconnect() except ValueError as e: logger.critical(f"{service_name}: 错误: {e}") except Exception as e: logger.critical(f"{service_name}: 未知错误: {e}") # MYSQL def mysql_partition_monitor(cursor_db, mysql_ip): try: #读取跳过表的列表 excluded_tables = set() try: with open('/home/oracle/skip_tables.list', 'r') as f: for line in f: line = line.strip() # 忽略空行和注释的行 if line and not line.startswith('#') and line.endswith('mysql'): parts = line.split(':') if len(parts) >= 3: ip, schema, table = parts[0], parts[1], parts[2] # 只处理当前ip if ip == mysql_ip: excluded_tables.add((schema, table)) except FileNotFoundError: # 查不到则使用默认配置 pass mysql_query_partition = """SELECT TABLE_SCHEMA, TABLE_NAME, MAX(PARTITION_ORDINAL_POSITION) as MAX_PARTITION_ORDINAL_POSITION FROM information_schema.partitions WHERE TABLE_SCHEMA not in ('mysql','sys','information_schema','performance_schema') AND PARTITION_METHOD is not null AND PARTITION_DESCRIPTION <> 'MAXVALUE' GROUP BY TABLE_SCHEMA, TABLE_NAME ORDER BY TABLE_SCHEMA, TABLE_NAME""" cursor_db.execute(mysql_query_partition) partitions = cursor_db.fetchall() if partitions is None or not partitions: logger.info(f"{mysql_ip}:不存在分区表") return None else: # 跳过无需巡检的分区表 filtered_patitions = [] for partition in partitions: schema_name = partition[0] table_name = partition[1] # 如果需要跳过，则跳过 if (schema_name, table_name) in excluded_tables: continue filtered_patitions.append(partition) for partition in filtered_patitions: detail_query = """SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, PARTITION_EXPRESSION, PARTITION_DESCRIPTION FROM information_schema.partitions WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s AND PARTITION_ORDINAL_POSITION=%s""" cursor_db.execute(detail_query, (partition[0], partition[1], partition[2])) detail_result = cursor_db.fetchall() for row_result in detail_result: TABLE_SCHEMA = row_result[0] TABLE_NAME = row_result[1] PARTITION_NAME = row_result[2] PARTITION_EXPRESSION = row_result[3] PARTITION_DESCRIPTION = row_result[4] formatted_partition_value = format_partition_description(PARTITION_EXPRESSION, PARTITION_DESCRIPTION, cursor_db) # 将格式化后的 PARTITION_DESCRIPTION 转换为整数以便比较 try: formatted_partition_value_int = int(formatted_partition_value) if formatted_partition_value_int < current_date_int: logger.error(f"{mysql_ip}:库名:{TABLE_SCHEMA},表名:{TABLE_NAME},目前最大分区为:{PARTITION_NAME},分区描述:{PARTITION_DESCRIPTION},分区最高范围至:{formatted_partition_value},分区已过期") elif formatted_partition_value_int - current_date_int <= 30: logger.warning(f"{mysql_ip}:库名:{TABLE_SCHEMA},表名:{TABLE_NAME},目前最大分区为:{PARTITION_NAME},分区描述:{PARTITION_DESCRIPTION},分区最高范围至:{formatted_partition_value},需及时建立新分区") else: logger.info(f"{mysql_ip}:库名:{TABLE_SCHEMA},表名:{TABLE_NAME},目前最大分区为:{PARTITION_NAME},分区描述:{PARTITION_DESCRIPTION},分区最高范围至:{formatted_partition_value},分区正常") except ValueError: formatted_partition_value_int = 0 # 如果无法转换为整数，则设为 0 return filtered_patitions except Exception as e: logger.error(f"{mysql_ip}:查询分区信息出错:{e}") return None # mysql可能存在的格式'2023-03-01 00:00:00','20240701',20230318,1754582400000,1727712000,'202501',202501,738764(to_days) def format_partition_description(PARTITION_EXPRESSION, partition_description, cursor_db): value_str = str(partition_description).strip() if 'to_days' not in str(PARTITION_EXPRESSION).lower(): date_pattern1 = re.compile(r"'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})'") match = date_pattern1.search(value_str) if match: try: date_str = match.group(1) date_obj = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S') return date_obj.strftime('%Y%m%d') except ValueError: pass date_pattern2 = re.compile(r"'(\d{8})'") match = date_pattern2.search(value_str) if match: return match.group(1) date_pattern3 = re.compile(r"^(\d{8})$") match = date_pattern3.search(value_str) if match: return match.group(1) date_pattern4 = re.compile(r"'(\d{6})'") match = date_pattern4.search(value_str) if match: date_str = match.group(1) return f"{date_str}01" date_pattern5 = re.compile(r"^(\d{6})$") match = date_pattern5.search(value_str) if match: date_str = match.group(1) return f"{date_str}01" date_pattern6 = re.compile(r"^(\d{9,})$") match = date_pattern6.search(value_str) if match: timestamp = match.group(1) try: if len(timestamp) >= 13: timestamp_seconds = int(timestamp[:10]) date_obj = datetime.fromtimestamp(timestamp_seconds) # print(date_obj.strftime('%Y-%m-%d %H:%M:%S')) return date_obj.strftime('%Y%m%d') elif len(timestamp) >= 10: timestamp_seconds = int(timestamp) date_obj = datetime.fromtimestamp(timestamp_seconds) # print(date_obj.strftime('%Y-%m-%d %H:%M:%S')) return date_obj.strftime('%Y%m%d') except (ValueError, OSError): # 如果时间戳解析失败，返回原值 return timestamp else: return value_str else: # 针对to_days函数做分区条件的转化 date_pattern7 = re.compile(r"(\d{6})") date_format_current = date_pattern7.search(value_str) if date_format_current: date_format_current_str = date_format_current.group(1) if len(date_format_current_str) == 6 and date_format_current_str.startswith('7'): try: sql = """select date_format(str_to_date(from_days(%s),'%%Y-%%m-%%d'),'%%Y%%m%%d') from dual""" cursor_db.execute(sql, (date_format_current_str,)) result = cursor_db.fetchone() if result: return result[0] else: return value_str except Exception as e: logger.error(f"执行 from_days 查询失败: {e}") return value_str else: return value_str # 密码通过特殊处理，若密码写死则不需要该方法 def get_mysql_password(mysql_ip, username): try: result = subprocess.run(['/home/oracle/bin/get_pswd.sh', mysql_ip, username], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) return result.stdout.strip() except subprocess.CalledProcessError as e: logger.critical(f"{mysql_ip}:获取MySQL密码失败: {e}") return None except FileNotFoundError: logger.critical("找不到get_user_pswd.sh脚本") return None #OB_MYSQL def ob_mysql_partition_monitor(cursor_ob_db, tenant_name): try: #读取跳过表的列表 excluded_tables = set() try: with open('/home/oracle/skip_tables.list', 'r') as f: for line in f: line = line.strip() # 忽略空行和注释的行 if line and not line.startswith('#') and line.endswith('oceanbase'): parts = line.split(':') if len(parts) >= 2: tenantname, schema, table = parts[0], parts[1], parts[2] if tenantname == tenant_name: excluded_tables.add((schema, table)) except FileNotFoundError: # 查不到则使用默认配置 pass query_partition = """SELECT TABLE_SCHEMA, TABLE_NAME, MAX(PARTITION_ORDINAL_POSITION) as MAX_PARTITION_ORDINAL_POSITION FROM information_schema.partitions WHERE TABLE_SCHEMA not in ('mysql','sys','information_schema','performance_schema') AND PARTITION_METHOD is not null AND PARTITION_DESCRIPTION <> 'MAXVALUE' GROUP BY TABLE_SCHEMA, TABLE_NAME ORDER BY TABLE_SCHEMA, TABLE_NAME""" cursor_ob_db.execute(query_partition) partitions = cursor_ob_db.fetchall() if partitions is None or not partitions: logger.info(f"{tenant_name}:不存在分区表") return None else: # 跳过无需巡检的分区表 filtered_patitions = [] for partition in partitions: schema_name = partition[0] table_name = partition[1] # 如果需要跳过，则跳过 if (schema_name, table_name) in excluded_tables: continue filtered_patitions.append(partition) for partition in filtered_patitions: detail_query = """SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, PARTITION_DESCRIPTION FROM information_schema.partitions WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s AND PARTITION_ORDINAL_POSITION=%s""" cursor_ob_db.execute(detail_query, (partition[0], partition[1], partition[2])) detail_result = cursor_ob_db.fetchall() for row_result in detail_result: OB_TABLE_SCHEMA = row_result[0] OB_TABLE_NAME = row_result[1] OB_PARTITION_NAME = row_result[2] OB_PARTITION_DESCRIPTION = row_result[3] ob_formatted_partition_value = ob_format_partition_description(OB_PARTITION_DESCRIPTION) # 将格式化后的 PARTITION_DESCRIPTION 转换为整数以便比较 try: ob_formatted_partition_value_int = int(ob_formatted_partition_value) if ob_formatted_partition_value_int < current_date_int: logger.error(f"{tenant_name}:库名:{OB_TABLE_SCHEMA},表名:{OB_TABLE_NAME},目前最大分区为:{OB_PARTITION_NAME},分区描述:{OB_PARTITION_DESCRIPTION},分区最高范围至:{ob_formatted_partition_value},分区已过期") elif ob_formatted_partition_value_int - current_date_int <= 30: logger.warning(f"{tenant_name}:库名:{OB_TABLE_SCHEMA},表名:{OB_TABLE_NAME},目前最大分区为:{OB_PARTITION_NAME},分区描述:{OB_PARTITION_DESCRIPTION},分区最高范围至:{ob_formatted_partition_value},需及时建立新分区") else: logger.info(f"{tenant_name}:库名:{OB_TABLE_SCHEMA},表名:{OB_TABLE_NAME},目前最大分区为:{OB_PARTITION_NAME},分区描述:{OB_PARTITION_DESCRIPTION},分区最高范围至:{ob_formatted_partition_value},分区正常") except ValueError: ob_formatted_partition_value_int = 0 # 如果无法转换为整数，则设为 0 return partitions except Exception as e: logger.error(f"{tenant_name}:查询分区信息出错:{e}") return None # ob-mysql可能存在的格式'2021-01-01 00:00:00','2021-01-01' def ob_format_partition_description(OB_PARTITION_DESCRIPTION): value_str = str(OB_PARTITION_DESCRIPTION).strip() date_pattern1 = re.compile(r"'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})'") match = date_pattern1.search(value_str) if match: try: date_str = match.group(1) date_obj = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S') return date_obj.strftime('%Y%m%d') except ValueError: pass date_pattern2 = re.compile(r"'(\d{4}-\d{2}-\d{2})'") match = date_pattern2.search(value_str) if match: date_obj = datetime.strptime(match.group(1), '%Y-%m-%d') return date_obj.strftime('%Y%m%d') else: return value_str if __name__ == '__main__': logger.info(f"分区巡检开始，当前日期: {current_date_str}") logger.info(f"分区巡检日志路径:{log_file}") service_names = ['adg_orcl1','adg_orcl2'] try: logger.info(f"Oracle分区巡检开始:") for service_name in service_names: ora_partition_monitor(service_name) except cx_Oracle.DatabaseError as e: logger.critical(f"{service_name}: {e}") except ValueError as e: logger.critical(f"{service_name}: {e}") try: logger.info(f"Mysql分区巡检开始:") # 此处的mysql_ip存储在oracle表中，也可以写死在文件当中，根据需求修改 zytj_db = ConnOracle('query', 'password', '带入ip', 1521, 'xjdb') mysql_ips_sql='''select HOST from MYSQL_HOST''' mysql_ips=zytj_db.select_noheader(mysql_ips_sql) if mysql_ips is None or len(mysql_ips) == 0: logger.critical("No MySQL IPs found in the database.") else: for row in mysql_ips: mysql_ip = row[0] try: # 此处密码通过特殊处理，也可以保持待巡检机器的账号密码相同，只要在脚本中写死即可 password = get_mysql_password(mysql_ip, 'query') db_config = { 'host': mysql_ip, 'port': 3306, 'user': 'query', 'password': password } conn_db = pymysql.connect(**db_config) cursor_db = conn_db.cursor() #获取分区信息 partitions = mysql_partition_monitor(cursor_db, mysql_ip) cursor_db.close() conn_db.close() except Exception as e: logger.error(f"MySQL连接或查询失败 {mysql_ip}: {e}") except ValueError as e: logger.critical(f"错误: {e}") except Exception as e: logger.critical(f"未知错误: {e}") try: logger.info(f"OB-Mysql分区巡检开始:") if tenant_info.returncode == 0: lines = tenant_info.stdout.strip().split('\n') mysql_tenants = [] for line in lines: if line.startswith('MYSQL') and not line.endswith('sys'): parts = line.split() if len(parts) >= 2: tenant_name = parts[1] mysql_tenants.append(tenant_name) logger.info(f"获取MYSQL租户:{mysql_tenants}") for tenant_name in mysql_tenants: try: # 根据实际情况带入租户名，提前在ocp建好巡检用户 username = f'xj_user@{tenant_name}#obtestprd' db_config = { 'host': 'ob_ip', 'port': 2883, 'user': username, 'password': 'password' } conn_ob_mysql = pymysql.connect(**db_config) cursor_ob_db = conn_ob_mysql.cursor() #获取分区信息 partitions = ob_mysql_partition_monitor(cursor_ob_db, tenant_name) cursor_ob_db.close() conn_ob_mysql.close() except Exception as e: logger.error(f"MySQL连接或查询失败 {tenant_name}: {e}") except ValueError as e: logger.critical(f"错误: {e}") except Exception as e: logger.critical(f"未知错误: {e}")