# -*- coding: utf-8 -*- # """ 1.输入表名，自动生成比对配置文件 2.执行sync-diff比对 3.记录比对日志 4.增加ignore cols属性 """ import random import string import os import re import subprocess import time from datetime import date, datetime from configparser import ConfigParser import ast from multiprocessing import Pool import mysql.connector # import insert_testdata def f_generate_sync_config(table_schema='', table_name='', ignore_cols='', conf_dir='/tmp' ): """ generate toml config file """ print("%s [INFO] start generate o2t-sync-diff config for %s.%s" % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), table_schema, table_name)) config_text = ''' # Sync Diff Configuration. ######################### Global config ######################### log-level = "info" chunk-size = 100000 check-thread-count = 10 sample-percent = 100 use-checksum = true only-use-checksum = true use-checkpoint = false ignore-data-check = false ignore-struct-check = true fix-sql-file = "./logs/%s.%s_fix.sql" #fix-target = "Oracle" ######################### Tables config ######################### [[check-tables]] schema = "%s" oracle-schema = "%s" tables = ["%s"] ######################### Databases config ######################### [[source-db]] type = "Oracle" user = "%s" password = "%s" connectString = "127.0.0.1:1521/oratidb?connect_timeout=2" instance-id = "source-1" #charset = "zhs16gbk" [target-db] host = "127.0.0.1" port = 4100 user = "" password = "" instance-id = "source-2" ''' %(table_schema, table_name, table_schema, table_schema, table_name, table_schema, table_schema) if ignore_cols == '': ignore_column_text = '' else: ignore_column_text = ''' ######################### Tables config ######################### [[table-config]] schema = "%s" table = "%s" ignore-columns = ["%s"] ''' %(table_schema, table_name, ignore_cols.replace(',', '","')) # print(config_text) config_name = '%s/%s_%s.toml' %(conf_dir, table_schema, table_name) # print(config_name) with open(config_name, 'w') as file: file.write(config_text+'\n'+ignore_column_text) return config_name def f_run_sync_diff(binary_path=".", config_file="", log_file=""): print("%s [INFO] start run o2t-sync-diff program,config=%s" % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), config_file)) shell_cmd = '%s/o2t_sync_diff -config %s > /dev/null ' %(binary_path, config_file) rt_code = os.system(shell_cmd) if rt_code == 0: print("%s [SUCC] run o2t-sync-diff success, table count %s, retrun code %s" % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), tableCount, rt_code)) rt_code = "compare_succ" else: print("%s [ERRO] run o2t-sync-diff failed,retrun code %s" % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), rt_code)) print("%s [INFO] CMD: %s" %(time.strftime("%Y-%m-%d-%H.%M.%S",time.localtime()), shell_cmd) ) rt_code = "compare_fail:%s" %(rt_code) return rt_code def f_get_table_list(dbconn=""): print("%s [INFO] function %s start" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), f_get_table_list.__name__)) #stmt_output = ["bs_intr_acsetl_detail2", "ogg_checkpoint_table", "YUJIE1123B", "ogg_checkpoint_table", "ogg_checkpoint_table_lox" ] global sync_tables_lst global ignore_cols_dict sync_tables_lst = [] ignore_cols_dict = {} stmt_query = """ select concat(table_schema,'.',table_name),ignore_columns from test.table_sync_diff_info t where sync_status='compare_fail:256' and ifnull(table_label,'')<>'all_cols_ignore' and length(ignore_columns)>1 order by table_name limit 20000 """ stmt_query = """ select concat(table_schema,'.',table_name),ifnull(ignore_columns,'') from test.table_sync_diff_info t where sync_status='waiting' order by table_schema,table_name limit 25000 """ # cnx = mysql.connector.connect(**dbconn) # cursor = cnx.cursor() cursor.execute(stmt_query) for row in cursor: sync_tables_lst.append(row[0]) if len(row[1]) > 1: ignore_cols_dict[row[0]] = row[1] print("find %s tables to compare, %s tables contain ignore column " % (len(sync_tables_lst), len(ignore_cols_dict))) # cursor.close() # cnx.close() time.sleep(2) return sync_tables_lst def run_sync_ctl(wrokerid=0, worker_count=10): print("%s [INFO] function %s start" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), run_sync_ctl.__name__)) # read config file #global config #global db_conn_dict #config = ConfigParser() #config.read('sync_dbcfg.ini') #db_conn_str = config.get('tidb', 'connstr') #db_conn_dict = ast.literal_eval(db_conn_str) #print('db_conn_dict:', type(db_conn_dict), db_conn_dict) #global batchid #batchid = time.strftime("%Y%m%d.%H%M%S", time.localtime()) workDir = os.path.dirname(os.path.abspath(__file__)) binDir = "/data2/autotest/localdrainer/oracle" confDir = "%s/config" % (workDir) logDir = "%s/logs" % (workDir) if not os.path.exists(confDir): os.makedirs(confDir) if not os.path.exists(logDir): os.makedirs(logDir) global cursor cnx = mysql.connector.connect(**db_conn_dict) cursor = cnx.cursor() tables_lst = f_get_table_list(dbconn=db_conn_dict) i = 0 for table in tables_lst: i = i + 1 if i % worker_count != wrokerid: #parallel continue else: pidremark = "%s-%s" % (os.getppid(), os.getpid()) print("%s [INFO] %s:%s:%s" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), pidremark, i, table)) time.sleep(random.randint(1, 10)/10) tableSchema = table.split('.')[0] tableName = table.split('.')[1] if table in ignore_cols_dict: ignore_column = ignore_cols_dict[table] else: ignore_column = '' confFile = f_generate_sync_config(table_name=tableName, table_schema=tableSchema, conf_dir=confDir, ignore_cols=ignore_column) # print(confFile, type(confFile)) # f_check_table_count(tab=tableName) # # #get table_count global tableCount stmt_select = "select count(1) from %s.%s " % (tableSchema, tableName) cursor.execute(stmt_select) for row in cursor: tableCount = row[0] sync_starttime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) t0 = time.time() sync_rt_code = f_run_sync_diff(binary_path=binDir, config_file=confFile, log_file=logDir) stmt_merge_log = """update test.table_sync_diff_info set batchid = '%s', table_count = %d, sync_status = '%s', sync_starttime = '%s', sync_endtime = now() , sync_duration = %d , remark = '%s' where table_schema='%s' and table_name='%s' """ % (batchid, tableCount, sync_rt_code, sync_starttime, time.time() - t0, pidremark, tableSchema, tableName) cursor.execute(stmt_merge_log) cnx.commit() stmt_hislog = """insert into test.table_sync_diff_info_his select t.*,now() from test.table_sync_diff_info t where table_schema='%s' and table_name='%s' """ % (tableSchema, tableName) cursor.execute(stmt_hislog) cnx.commit() cursor.close() cnx.close() if __name__ == '__main__': print('%s [INFO] I am child process (%s) and my parent is %s.' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), os.getpid(), os.getppid())) print('%s [INFO] Waiting for all subprocesses start...' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()))) global config global db_conn_dict config = ConfigParser() config.read('sync_dbcfg.ini') db_conn_str = config.get('tidb', 'connstr') db_conn_dict = ast.literal_eval(db_conn_str) print('db_conn_dict:', type(db_conn_dict), db_conn_dict) global batchid batchid = time.strftime("%Y%m%d.%H%M%S", time.localtime()) p_num = 5 p = Pool(p_num) for i in range(p_num): p.apply_async(run_sync_ctl, args=(i, p_num)) #run_sync_ctl(1,2) print('%s [INFO] Waiting for all subprocesses done...' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()))) p.close() p.join() print('%s [INFO] All subprocesses done.' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()))) #analyze result from analyze_syncdiff_result import f_get_syncdiff_data print("batchid:",batchid,type(batchid)) input_batchid = "'%s'"%(batchid) f_get_syncdiff_data(db_conn_str, input_batchid)