皮蛋壳 - 万恶的皮蛋
type: channel
link: https://wedpd.cn/index.php/author/2/
description: 万恶的皮蛋
items: Oceanbase Mysql租户的统计信息收集脚本
items: MySql binlog大事务巡检脚本
items: Mysql+Oracle+OceanBase分区表Python巡检脚本

Oceanbase Mysql租户的统计信息收集脚本
type: item
date: 2025-10-15T01:58:00+00:00
description: 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_nam…

MySql binlog大事务巡检脚本
type: item
date: 2025-10-15T01:57:00+00:00
description: 此脚本支持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}'" c…

Mysql+Oracle+OceanBase分区表Python巡检脚本
type: item
date: 2025-10-15T01:57:00+00:00
description: {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:…