from multiprocessing import Process from multiprocessing import Pool import logging import time import pymysql import os import requests import json ######## 对所有库重要表进行analyze，循环运行，中间停顿15分钟 ######### 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_frequently_less80_5500-5599_%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_55%';" cursor.execute(db_sql) row_all = cursor.fetchall() important_db_list=[] db_list=[] with open("/home/tidb/scripts/analyze_stats/site_config.txt", "r") as file: for line in file: site = line.strip() important_db_list.append(site) with open("/home/tidb/scripts/analyze_stats/promotion_site_config.txt", "r") as file: for line in file: site = line.strip() important_db_list.append(site) for i in row_all: if i[0] not in important_db_list: db_list.append(i[0]) conn.commit() cursor.close() conn.close() return sorted(db_list,reverse=True) 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_55%';" cursor.execute(db_sql) row_all = cursor.fetchall() conn.commit() cursor.close() conn.close() return row_all def make_healthy_info(healthy_info,table_list): healthy_dic={} for healthy in healthy_info: healthy_less_dic.setdefault(healthy[0],[]).append(healthy[1]) return healthy_dic def healthy_less(healthy_info,table_list): healthy_less_list=[] for healthy in healthy_info: if healthy[1] in table_list and healthy[3]<60: healthy_less_list.append([healthy[0],healthy[1],healthy[3]]) return healthy_less_list def group_healthy_less(healthy_less_list): l=[] table_info="" table_dic={} for x in healthy_less_list: l.append(x[1]) sl = set(l) for i in sl: table_dic[i]=l.count(i) print(table_dic.items()) a=sorted(table_dic.items(), key = lambda kv:(kv[1], kv[0]),reverse=True) for b,c in a: table_info +="{}:{}\n".format(b,c) return table_info 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] <= 3: break elif row_all[0][0] > 3: 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 with 256 buckets, 500 topn;" % (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.debug("%s-%s less %s" % (dh_db,table,healthy_stat)) 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]<60: analyze(dh_db,table,healthy_table[3]) if __name__ == "__main__": while True: now=time.strftime("%Y%m%d %H:%M:%S", time.localtime()) logging.info("脚本开始时间 %s\n" % (now)) print('Parent process %s.' % os.getpid()) p = Pool(5) healthy_info=get_healthy_info() site=get_site() print(site) table_list=[] for table in open("/home/tidb/scripts/analyze_stats/frequently_table.txt"): table_list.append(table.strip("\n").strip()) healthy_less_list=healthy_less(healthy_info,table_list) group_info=group_healthy_less(healthy_less_list) for site_code in site: if site_code != "dh_app_default" : p.apply_async(make_analyze, args=(site_code,healthy_info,table_list,)) 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 5500-5599库重要表健康度小于80的统计信息收集，开始时间: %s 结束时间: %s. 健康度小于80的重要表为%d %s""" % (now,end_time,len(healthy_less_list),group_info) logging.info("脚本执行信息 %s\n" % (msg)) webhook(msg) print('All subprocesses done.') time.sleep(300)