# -*- coding: utf-8 -*- # """ 1.根据配置文件的测试datatype范围，去columns表里找包含此类型的表 2.找到的表名，在源端tidb逐个表依次执行truncate(delete)+insert...select... 3.执行完成后，在目标端oracle查询数据量总数目 4.将各个步骤的开始结束时间、数据变动量写入日志表 5.遇到复制报错导致drainer停止的，自动清理checkpoint和拉起drainer """ import random import string import os import re import subprocess import time import datetime from configparser import ConfigParser import ast import cx_Oracle import mysql.connector def exetime(func): def newFunc(*args, **args2): t0 = time.time() print ("@%s, {%s} start" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), func.__name__)) func(*args, **args2) print ("@%s, {%s} end" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), func.__name__)) print ("@%.3fs taken for {%s}" % (time.time() - t0, func.__name__)) return newFunc @exetime def f_get_table_list(): """ get table list from tidb """ config = ConfigParser() config.read('dbcfg.ini') global tab_schema global conn_str db_conn_dict = ast.literal_eval(config.get('tidb', 'connstr')) conn_str = db_conn_dict #get config data type table_contain_datatype = config.get('table', 'table_contain_datatype').replace(' ','') table_notcontain_datatype = config.get('table', 'table_notcontain_datatype').replace(' ','') table_contain_pkuk = config.get('table', 'table_contain_pkuk') tab_schema = config.get('db', 'dbschema') # ast.literal_eval( datatype_in_lst = table_contain_datatype.split(',') datatype_notin_lst = table_notcontain_datatype.split(',') print(type(datatype_in_lst), datatype_in_lst) print(type(datatype_notin_lst), datatype_notin_lst) if table_contain_pkuk.lower() == "yes": stmt_pkflag = """ and exists (select 1 from INFORMATION_SCHEMA.table_constraints t4 where t1.table_schema=t4.table_schema and t1.table_name=t4.table_name and t4.constraint_type in ('PRIMARY KEY', 'UNIQUE') )""" elif table_contain_pkuk.lower() == "no": stmt_pkflag = """ and not exists (select 1 from INFORMATION_SCHEMA.table_constraints t4 where t1.table_schema=t4.table_schema and t1.table_name=t4.table_name and t4.constraint_type in ('PRIMARY KEY', 'UNIQUE') )""" else: stmt_pkflag = "" stmt_query = """ select table_name from INFORMATION_SCHEMA.tables t1 where table_schema='%s' and exists (select 1 from INFORMATION_SCHEMA.columns t2 where t1.table_schema=t2.table_schema and t1.table_name=t2.table_name and t2.data_type in ('%s') ) and not exists (select 1 from INFORMATION_SCHEMA.columns t3 where t1.table_schema=t3.table_schema and t1.table_name=t3.table_name and t3.data_type in ('%s') ) and not exists (select 1 from INFORMATION_SCHEMA.columns t0 where t1.table_schema=t0.table_schema and t1.table_name=t0.table_name and t0.extra in ('STORED GENERATED','VIRTUAL GENERATED') ) %s order by table_name """ % (tab_schema, "','".join(datatype_in_lst).lower(), "','".join(datatype_notin_lst).lower(), stmt_pkflag) print(stmt_query) global stmt_output stmt_output = [] # ignore table , already replicate succ ignore_tab_set = set() ignore_tab_file = "ignore_tabs.txt" try: f = open(ignore_tab_file, mode='r') for line in f: ignore_tab_set.add(line.split()[0]) f.close() print("total %s tables in ignore list" % (len(ignore_tab_set))) except FileNotFoundError: print("no table in ignore list %s" %(ignore_tab_file)) cnx = mysql.connector.connect(**conn_str) cursor = cnx.cursor() cursor.execute(stmt_query) for row in cursor: if row[0] not in ignore_tab_set: stmt_output.append(row[0]) print("find %s tables to insert data" % (len(stmt_output))) cursor.close() cnx.close() time.sleep(2) # print(stmt_output) def f_insert_data(): """ tidb insert """ source_schema = "%s_raw" %(tab_schema) target_schema = tab_schema i = 0 batchid = time.strftime("%Y%m%d.%H%M%S", time.localtime()) cnx_o = cx_Oracle.connect(user="", password="", dsn="127.0.0.1:1521/oratidb") cursor_o = cnx_o.cursor() cnx = mysql.connector.connect(**conn_str) cursor = cnx.cursor() for tab in stmt_output: stmt_select = "select count(1) from %s.%s " % (target_schema, tab) stmt_delete = "truncate table %s.%s " % (target_schema, tab) stmt_insert = "insert into %s.%s select * from %s.%s" % (target_schema, tab, source_schema, tab) # print("%s\n%s" %(stmt_delete, stmt_insert)) # print("start %s.%s" %(target_schema, tab)) i = i + 1 rows_delete = 0 rows_insert = 0 run_failed = 0 if i > 10: break #for test try: cursor.execute(stmt_delete) rows_delete = cursor.rowcount insert_starttime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) cursor.execute(stmt_insert) rows_insert = cursor.rowcount print("%s [SUCC] execute insert finish, %s %s(delete) %s(insert) rows affect, the %s table" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), tab, rows_delete, rows_insert, i)) cnx.commit() repl_status='insert_succ' except: print("%s [ERRO] insert failed,exit %s" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), stmt_insert)) run_failed = run_failed + 1 repl_status='insert_fail' # exit() stmt_merge_log = """update test.table_repl_info set batchid = '%s', repl_status = '%s', source_count = '%s', insert_starttime = '%s', insert_endtime = now() , repl_starttime = now() where table_schema='%s' and table_name='%s' """ %(batchid, repl_status, rows_insert, insert_starttime, target_schema, tab) cursor.execute(stmt_merge_log) cnx.commit() if rows_insert == 0: continue target_rows = 0 repl_status = 'repl_timeout' ##check Oracle Replicate speed for loop_i in range(60 + int(rows_insert/1000)): try: cursor_o.execute(stmt_select) except cx_Oracle.DatabaseError as e: if error.code == 8103: # select return "ORA-08103: object no longer exists" when truncate print('Database error: %s. retry...'.format(e)) continue else: print('Database error: %s'.format(e)) for row in cursor_o: target_rows = row[0] if target_rows == rows_insert: repl_status = 'replicate_succ' print("%s [SUCC] target insert finish, %s %s(replicate) rows affect" % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), tab, target_rows)) break time.sleep(1) ##check drainer status #drop checkpoint table and restart drainer if repl_status == 'repl_timeout': print("%s [WARN] replicate may failed, target rows %s, start check process..." % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), target_rows)) ps_str = os.popen("ps -ef | grep dr_ora | grep -v grep | wc -l").read() if int(ps_str) == 0: print("%s [ERRO] drainer process is down, try restart " % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()))) repl_status = 'replicate_exit' # delete checkpoint delete_checkpoint = "delete from findpt_lj.TIDB_BINLOG_CHECKPOINT" cursor_o.execute(delete_checkpoint) print("%s [INFO] clear findpt_lj.TIDB_BINLOG_CHECKPOINT, %s rows affect " % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), cursor_o.rowcount)) cnx_o.commit() # print("%s [INFO] start drainer background " % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()))) os.system('nohup sh /data1/tmp/localdrainer/oracle/drainer -config dr_ora.toml &') #wait drainer start work time.sleep(300) # exit() else: ps_str = os.popen("ps -ef | grep dr_ | grep -v grep ").read() ps_lst = re.split(r"[ ]+", ps_str) print("%s [INFO] drainer is running from %s" % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), ps_lst[4])) #update replicate time stmt_merge_log = """update test.table_repl_info set repl_status = '%s', target_count = '%s', repl_endtime = now() , repl_duration = TIMESTAMPDIFF(SECOND,repl_starttime, now()) where table_schema='%s' and table_name='%s' """ %(repl_status, target_rows, target_schema, tab) cursor.execute(stmt_merge_log) cnx.commit() if repl_status == 'replicate_exit': print("%s [ERRO] replicate may failed %s, pls check drainer status" % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), tab)) exit() cursor.close() cnx.close() cursor_o.close() cnx_o.close() def main(): """ """ #run test result global config config = ConfigParser() config.read('dbcfg.ini') # read config file db_conn_dict = ast.literal_eval(config.get('tidb', 'connstr')) conn_str = db_conn_dict print(type(conn_str), conn_str) f_get_table_list() f_insert_data() if __name__ == '__main__': main()