from multiprocessing import Process from operator import itemgetter from multiprocessing import Pool import logging import time import pymysql import os import requests import json #### 对所有站点，次重要表进行analyze ######## api = 'https://zoom-api.cg.app/zoom/robotSendMsg' headers = {'Content-type': 'application/json', 'Accept': 'text/plain', 'apiKey':'4sdW06G6EbuahmPioVmr12'} addr="privatelink-tidb.izmbtw11ilxn.clusters.tidb-cloud.com" user="analyze_user" passwd="7D94q9CptRLCmLzQYnZf" date=time.strftime("%Y%m%d%H%M%S", time.localtime()) now=time.strftime("%Y%m%d %H:%M:%S", time.localtime()) print(date) log="/home/tidb/scripts/analyze_stats/analyze_logs/analyze_secondary_table_less20_loop_%s.log" % (date) logging.basicConfig(filename=log, level=logging.DEBUG) print(log) # 重要库的列表，如果有重要库可以加入里面 #数据库连接 def db_conn(addr,user,passwd,db): mydb = pymysql.connect( host= addr, port=4000, user=user, passwd=passwd, db=db ) return mydb #发送成功信息��zoom def webhook(msg): send_data = { "to_jid": "bd6d6ad01baf43b3825853868f2c8202@conference.xmpp.zoom.us", "type": "message", "msg": msg } try: r = requests.post(url=api, headers=headers, json=send_data) if r.status_code == 200: print("send success") except Exception as e: print(e) print("bot send message err") #获取站点名称 def get_site(): conn = db_conn(addr,user,passwd,"information_schema") cursor = conn.cursor() db_sql="show databases like 'dh_app_%';" cursor.execute(db_sql) row_all = cursor.fetchall() row_all = sorted(row_all, reverse=True) conn.commit() cursor.close() conn.close() return row_all def get_healthy_info(): conn = db_conn(addr,user,passwd,"information_schema") cursor = conn.cursor() cursor.execute("set session tidb_multi_statement_mode='on';") cursor.execute("set session max_execution_time=300000;") db_sql="SHOW STATS_HEALTHY where db_name like 'dh_app_%' and db_name != 'dh_app_default';" cursor.execute(db_sql) row_all = cursor.fetchall() row_all = sorted(row_all, key=itemgetter(3)) conn.commit() cursor.close() conn.close() return row_all def get_tables(frequently_tables): conn = db_conn(addr,user,passwd,"dh_app_default") cursor = conn.cursor() cursor.execute("set session tidb_multi_statement_mode='on';") cursor.execute("set session max_execution_time=300000;") db_sql="show tables;" cursor.execute(db_sql) row_all = cursor.fetchall() table_list=[] for i in row_all: if i[0] not in frequently_tables: table_list.append(i[0]) conn.commit() cursor.close() conn.close() return table_list def analyze(dh_db,table,healthy_stat): while True: conn = db_conn(addr,user,passwd,"information_schema") cursor = conn.cursor() db_sql="select count(1) from PROCESSLIST where info like 'analyze%' limit 1;" cursor.execute(db_sql) row_all = cursor.fetchall() if row_all[0][0] == 0: break elif row_all[0][0] > 0: cursor.close() conn.close() time.sleep(5) continue cursor.execute("set session tidb_multi_statement_mode='on';") cursor.execute("set session tidb_build_stats_concurrency=30;") cursor.execute("set session tidb_index_serial_scan_concurrency=10;") cursor.execute("set session tidb_distsql_scan_concurrency=30;") cursor.execute("show variables like 'tidb_build_stats_concurrency';") analyze_sql="analyze table %s.%s;" % (dh_db,table) start_time=time.strftime("%Y%m%d %H:%M:%S", time.localtime()) cursor.execute(analyze_sql) end_time=time.strftime("%Y%m%d %H:%M:%S", time.localtime()) logging.info("开始时间: %s %s-%s healthy is %s " % (start_time,dh_db,table,healthy_stat)) logging.info("结束时间: %s %s %s is finished\n" % (end_time,dh_db,table)) cursor.close() conn.close() def make_analyze(dh_db,healthy_info,table_list): logging.debug("%s is start" % (dh_db)) healthy_table_list=[] healthy_table_name_list=[] for healthy in healthy_info: if dh_db==healthy[0]: healthy_table_list.append(healthy) healthy_table_name_list.append(healthy[1]) for table in table_list: table=table.strip("\n").strip() if table not in healthy_table_name_list: analyze(dh_db,table,'NULL') else: for healthy_table in healthy_table_list: if table==healthy_table[1] and healthy_table[3]<=20: analyze(dh_db,table,healthy_table[3]) if __name__ == "__main__": while True: logging.info("脚本开始时间 %s\n" % (now)) print('Parent process %s.' % os.getpid()) p = Pool(5) healthy_info=get_healthy_info() site=get_site() frequently_tables_list=[] for table in open("/home/tidb/scripts/analyze_stats/frequently_table.txt"): frequently_tables_list.append(table.strip("\n").strip()) table_list=get_tables(frequently_tables_list) processed_libraries = 0 library_counter = 0 """ processed_libraries 用于记录已处理的库的数量，library_counter 用于计算每处理完20个库后的重置操作。当 library_counter 达到20时，会将其重置为0，并检查 processed_li braries 是否也达到20，如果是，则重新获取 healthy_info。然后，代码会继续处理下一个20个库。这样，每处理完20个库，就会重新获取一次 healthy_info。 """ for site_code in site: logging.info("脚本执行信息：%s" % (healthy_info[0][0])) if site_code[0] != "dh_app_default" : p.apply_async(make_analyze, args=(site_code[0],healthy_info,table_list,)) processed_libraries += 1 library_counter += 1 if library_counter >= 20: library_counter = 0 if processed_libraries >= 20: processed_libraries = 0 healthy_info = get_healthy_info() print('Waiting for all subprocesses done...') p.close() p.join() logging.info("脚本结束时间 %s\n" % (time.strftime("%Y%m%d %H:%M:%S", time.localtime()))) end_time=time.strftime("%Y%m%d %H:%M:%S", time.localtime()) msg="北美saas所有库次重要表健康度小于20的统计信息收集，开始时间: %s 结束时间: %s" % (now,end_time) logging.info("脚本执行信息：%s" % (msg)) webhook(msg) print('All subprocesses done.') time.sleep(1200)