diff --git a/binlog2sql/binlog2sql.py b/binlog2sql/binlog2sql.py index 971ec55..37e4e0f 100755 --- a/binlog2sql/binlog2sql.py +++ b/binlog2sql/binlog2sql.py @@ -8,24 +8,27 @@ from pymysqlreplication.event import QueryEvent, RotateEvent, FormatDescriptionEvent from binlog2sql_util import command_line_args, concat_sql_from_binlog_event, create_unique_file, temp_open, \ reversed_lines, is_dml_event, event_type +from shutil import copyfile +import argparse +import re class Binlog2sql(object): def __init__(self, connection_settings, start_file=None, start_pos=None, end_file=None, end_pos=None, start_time=None, stop_time=None, only_schemas=None, only_tables=None, no_pk=False, - flashback=False, stop_never=False, back_interval=1.0, only_dml=True, sql_type=None): + flashback=False, stop_never=False, back_interval=1.0, only_dml=True, sql_type=None, save_as=None): """ conn_setting: {'host': 127.0.0.1, 'port': 3306, 'user': user, 'passwd': passwd, 'charset': 'utf8'} """ - if not start_file: - raise ValueError('Lack of parameter: start_file') + # if not start_file: + # raise ValueError('Lack of parameter: start_file') self.conn_setting = connection_settings - self.start_file = start_file + # self.start_file = start_file self.start_pos = start_pos if start_pos else 4 # use binlog v4 - self.end_file = end_file if end_file else start_file + # self.end_file = end_file if end_file else start_file self.end_pos = end_pos if start_time: self.start_time = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S") @@ -44,13 +47,50 @@ def __init__(self, connection_settings, start_file=None, start_pos=None, end_fil self.binlogList = [] self.connection = pymysql.connect(**self.conn_setting) + self.last_pos = self.end_pos + self.save_as = save_as + with self.connection as cursor: + + # 处理一下数据库和表 + if only_schemas: + cursor.execute('select SCHEMA_NAME from information_schema.SCHEMATA') + allschemas = [r[0] for r in cursor.fetchall()] + schemas = [] + re_sch = re.compile(only_schemas) + for s in allschemas: + if re_sch.match(s): + schemas.append(s) + if len(schemas): + self.only_schemas = schemas + else: + raise ValueError('指定的数据库不存在：%s' % only_schemas) + + if only_tables: + cursor.execute("select table_name from information_schema.tables where table_type='base table'") + alltables = [r[0] for r in cursor.fetchall()] + tables = [] + print(only_tables) + re_tb = re.compile(only_tables) + for s in alltables: + if re_tb.match(s): + tables.append(s) + if len(tables): + self.only_tables = tables + else: + raise ValueError('指定的库表不存在：%s' % only_tables) + cursor.execute("SHOW MASTER STATUS") self.eof_file, self.eof_pos = cursor.fetchone()[:2] cursor.execute("SHOW MASTER LOGS") bin_index = [row[0] for row in cursor.fetchall()] + + self.start_file = start_file if start_file else bin_index[0] + self.end_file = end_file if end_file else bin_index[-1] + if self.start_file not in bin_index: raise ValueError('parameter error: start_file %s not in mysql server' % self.start_file) + binlog2i = lambda x: x.split('.')[1] for binary in bin_index: if binlog2i(self.start_file) <= binlog2i(binary) <= binlog2i(self.end_file): @@ -67,10 +107,11 @@ def process_binlog(self): only_tables=self.only_tables, resume_stream=True, blocking=True) flag_last_event = False - e_start_pos, last_pos = stream.log_pos, stream.log_pos + e_start_pos, self.last_pos = stream.log_pos, stream.log_pos # to simplify code, we do not use flock for tmp_file. tmp_file = create_unique_file('%s.%s' % (self.conn_setting['host'], self.conn_setting['port'])) - with temp_open(tmp_file, "w") as f_tmp, self.connection as cursor: + tmp_file_sql = create_unique_file('%s.%s.sql' % (self.conn_setting['host'], self.conn_setting['port'])) + with temp_open(tmp_file, "w") as f_tmp, temp_open(tmp_file_sql, "w") as f_tmp_sql, self.connection as cursor: for binlog_event in stream: if not self.stop_never: try: @@ -83,7 +124,7 @@ def process_binlog(self): elif event_time < self.start_time: if not (isinstance(binlog_event, RotateEvent) or isinstance(binlog_event, FormatDescriptionEvent)): - last_pos = binlog_event.packet.log_pos + self.last_pos = binlog_event.packet.log_pos continue elif (stream.log_file not in self.binlogList) or \ (self.end_pos and stream.log_file == self.end_file and stream.log_pos > self.end_pos) or \ @@ -94,13 +135,14 @@ def process_binlog(self): # raise ValueError('unknown binlog file or position') if isinstance(binlog_event, QueryEvent) and binlog_event.query == 'BEGIN': - e_start_pos = last_pos + e_start_pos = self.last_pos if isinstance(binlog_event, QueryEvent) and not self.only_dml: sql = concat_sql_from_binlog_event(cursor=cursor, binlog_event=binlog_event, flashback=self.flashback, no_pk=self.no_pk) if sql: - print(sql) + # print(sql) + f_tmp_sql.write(sql + '\n') elif is_dml_event(binlog_event) and event_type(binlog_event) in self.sql_type: for row in binlog_event.rows: sql = concat_sql_from_binlog_event(cursor=cursor, binlog_event=binlog_event, no_pk=self.no_pk, @@ -108,17 +150,22 @@ def process_binlog(self): if self.flashback: f_tmp.write(sql + '\n') else: - print(sql) + # print(sql) + f_tmp_sql.write(sql + '\n') if not (isinstance(binlog_event, RotateEvent) or isinstance(binlog_event, FormatDescriptionEvent)): - last_pos = binlog_event.packet.log_pos + self.last_pos = binlog_event.packet.log_pos if flag_last_event: break stream.close() f_tmp.close() + f_tmp_sql.close() + if self.flashback: self.print_rollback_sql(filename=tmp_file) + if self.save_as: + copyfile(tmp_file_sql, self.save_as) return True def print_rollback_sql(self, filename): @@ -138,13 +185,32 @@ def print_rollback_sql(self, filename): def __del__(self): pass - -if __name__ == '__main__': - args = command_line_args(sys.argv[1:]) - conn_setting = {'host': args.host, 'port': args.port, 'user': args.user, 'passwd': args.password, 'charset': 'utf8'} +def createSql(conf): + conn_setting = {'host': conf['host'], 'port': conf.getint('port'), 'user': conf['user'], 'passwd': conf['password'], 'charset': 'utf8'} + # 获得文件和最后的位置 + args = argparse.Namespace(back_interval=1.0, databases=conf['databases'], end_file='', end_pos=0, flashback=False, help=False, host='', no_pk=False, only_dml=False, password='', port=3306, save_as='', sql_type=['INSERT', 'UPDATE', 'DELETE'], start_file=None, start_pos=4, start_time='', stop_never=False, stop_time='', tables=conf['tables'], user='') binlog2sql = Binlog2sql(connection_settings=conn_setting, start_file=args.start_file, start_pos=args.start_pos, end_file=args.end_file, end_pos=args.end_pos, start_time=args.start_time, stop_time=args.stop_time, only_schemas=args.databases, only_tables=args.tables, no_pk=args.no_pk, flashback=args.flashback, stop_never=args.stop_never, - back_interval=args.back_interval, only_dml=args.only_dml, sql_type=args.sql_type) - binlog2sql.process_binlog() + back_interval=args.back_interval, only_dml=args.only_dml, sql_type=args.sql_type, save_as=args.save_as) + # 对比最后的位置是否有变化 + if conf.getint('position') != binlog2sql.eof_pos: + binlog2sql.save_as = "%s%s.%s.%s" % (conf['sqlFilePath'], conn_setting['host'], conn_setting['port'], conf['fileId']) + binlog2sql.start_file = conf['binlogfile'] + binlog2sql.start_pos = conf.getint('position') + + binlog2sql.process_binlog() + fileid = conf.getint('fileId') + 1 + + conf['fileId'] = str(fileid) + conf['position'] = str(binlog2sql.last_pos) + conf['binlogfile'] = binlog2sql.end_file + return { + 'sqlFile': binlog2sql.save_as, + 'PFileId': fileid, + 'position': binlog2sql.last_pos, + 'binlogfile': binlog2sql.end_file + } + else: + return None diff --git a/binlog2sql/binlog2sql_util.py b/binlog2sql/binlog2sql_util.py index 3a9bb31..a6fe155 100755 --- a/binlog2sql/binlog2sql_util.py +++ b/binlog2sql/binlog2sql_util.py @@ -76,6 +76,8 @@ def parse_args(): help="Start time. format %%Y-%%m-%%d %%H:%%M:%%S", default='') interval.add_argument('--stop-datetime', dest='stop_time', type=str, help="Stop Time. format %%Y-%%m-%%d %%H:%%M:%%S;", default='') + interval.add_argument('--save-as', dest='save_as', type=str, help='Save as file name', default='') + parser.add_argument('--stop-never', dest='stop_never', action="iframe.php?url=https%3A%2F%2Fgithub.com%2Fstore_true", default=False, help="Continuously parse binlog. default: stop at the latest event when you start.") parser.add_argument('--help', dest='help', action="iframe.php?url=https%3A%2F%2Fgithub.com%2Fstore_true", help='help information', default=False) @@ -109,8 +111,8 @@ def command_line_args(args): if args.help or need_print_help: parser.print_help() sys.exit(1) - if not args.start_file: - raise ValueError('Lack of parameter: start_file') + # if not args.start_file: + # raise ValueError('Lack of parameter: start_file') if args.flashback and args.stop_never: raise ValueError('Only one of flashback or stop-never can be True') if args.flashback and args.no_pk: @@ -181,7 +183,7 @@ def concat_sql_from_binlog_event(cursor, binlog_event, row=None, e_start_pos=Non elif flashback is False and isinstance(binlog_event, QueryEvent) and binlog_event.query != 'BEGIN' \ and binlog_event.query != 'COMMIT': if binlog_event.schema: - sql = 'USE {0};\n'.format(binlog_event.schema) + sql = 'USE {0};\n'.format(fix_object(binlog_event.schema)) sql += '{0};'.format(fix_object(binlog_event.query)) return sql @@ -265,3 +267,18 @@ def reversed_blocks(fin, block_size=4096): here -= delta fin.seek(here, os.SEEK_SET) yield fin.read(delta) + +def read_log(log_file): + if os.path.exists(log_file): + f = open(log_file, 'r') + return f.read().split('#') + else: + return [1,4,""] + +def write_log(log_file, content): + with open(log_file, 'w') as f: + f.write(content) + +if __name__ == "__main__": + write_log('log_12', "23#3434") + print(read_log('log_12')) diff --git a/binlog2sql/config.ini b/binlog2sql/config.ini new file mode 100644 index 0000000..cd3d705 --- /dev/null +++ b/binlog2sql/config.ini @@ -0,0 +1,22 @@ +[Producer] +host = 192.168.100.26 +port = 3306 +user = datacopy +password = passwd +tables = ^(tprj|tsys_enum).*$ +databases = security +sqlfilepath = /home/tias/dbsync/sql/ +sqlfilebakpath = /home/tias/dbsync/sql_bak/ +rsyncdest = rsync://192.168.100.28/dbsync/sql +fileid = 8 +position = 8604958 +binlogfile = mysql-bin.000002 + +[Consumer] +host = 192.168.100.28 +port = 3306 +user = tidyinfo +password = passwd +sqlfilepath = /home/tias/dbsync/sql/ +sqlfilebakpath = /home/tias/dbsync/sql_bak/ + diff --git a/binlog2sql/tiasMySqlSync.py b/binlog2sql/tiasMySqlSync.py new file mode 100644 index 0000000..8c26d91 --- /dev/null +++ b/binlog2sql/tiasMySqlSync.py @@ -0,0 +1,131 @@ +# 一个程序两种模式，备份模式，和导入模式 +# 备份模式，每分钟调用一次导出 sql，将文件存储在同步目录，然后调用同步命令同步文件，完成之后，移动文件到备份目录，结束 +# 导入模式，每分钟扫码一次同步目录，将文件安装顺序执行，完成后将文件移动到以处理目录 +# 如果出错需要发出警报，并且还需要记录处理日志 +# 如果处理出错需要停止，以及需要等待文件的顺序，所以各自需要记录自己的的文件编号指针，一单发现文件不是顺序的立即停止，等等，如果超过10分钟仍然未获取到文件，则报错 +import sys, os +from binlog2sql import createSql +import configparser + +if __name__ == "__main__": + configFile = "%s/config.ini" % os.path.split(os.path.realpath(sys.argv[0]))[0] + if not os.path.exists(configFile): + print("没找到配置文件，请确认，配置文件路径应该是:", configFile) + exit(1) + + config = configparser.ConfigParser() + config.read(configFile) + + mode = "p" if len(sys.argv) == 1 else sys.argv[1] + debug = True if len(sys.argv) < 3 else sys.argv[2]=='debug' + + ## 如果是生产者 + if mode == "p" or mode == "producer": + conf = config['Producer'] + + if not os.path.exists(conf['sqlFilePath']): + os.makedirs(conf['sqlFilePath']) + + if not os.path.exists(conf['sqlFileBakPath']): + os.makedirs(conf['sqlFileBakPath']) + + if createSql(conf): + # 记录日志 + if debug: + print("设置配置 fileId:%s\nposision:%s\nbinlogfile:%s" % (conf['fileId'],conf['position'],conf['binlogfile'])) + else: + print("保存 设置配置 fileId:%s\nposision:%s\nbinlogfile:%s" % (conf['fileId'],conf['position'],conf['binlogfile'])) + with open(configFile, 'w') as c: + config.write(c) + #同步文件 + cmd = "rsync -aPzv %s %s" % (conf['sqlFilePath'], conf['rsyncDest']) + if debug: + print("同步文件命令：", cmd) + result = 0 + else: + print("执行 同步文件命令：", cmd) + result = os.system(cmd) + + if result == 0: + print("同步文件完成") + cmd = "mv -f %s* %s" % (conf['sqlFilePath'], conf['sqlFileBakPath']) + if debug: + print("移除文件命令：", cmd) + else: + print("执行 移除文件命令：", cmd) + os.system(cmd) + + print("文件备份完成") + else: + print("同步文件失败") + + elif mode == "c" or mode == "consumer": + ## 如果是执行的一段 + ## 读取 + conf = config['Consumer'] + if not os.path.exists(conf['sqlFilePath']): + os.makedirs(conf['sqlFilePath']) + + if not os.path.exists(conf['sqlFileBakPath']): + os.makedirs(conf['sqlFileBakPath']) + + files= os.listdir(conf['sqlFilePath']) #得到文件夹下的所有文件名称 + sqls = [] + for file in files: #遍历文件夹 + if not os.path.isdir(file): #判断是否是文件夹，不是文件夹才打开 + sqls.append(file) + + if len(sqls) > 0: + sqls.sort(key=lambda x: int(x.split(".")[-1])) + logIndex = None ## 应该从上次的记录中获取 + index = conf.getint('fileId') if conf.getint('fileId') else 0 #logIndex if logIndex else (int(sqls[0].split(".")[-1])-1) + successSql = [] + for sql in sqls: + sqlid = int(sql.split(".")[-1]) + expectIndex = index + 1 + if expectIndex == sqlid: # 为了确保连续性 + # 可以执行文件 + cmd = "mysql -h%s -u%s -p%s < %s%s" % (conf['host'], conf['user'], conf['password'], conf['sqlFilePath'], sql) + if debug: + print("数据导入命令:", cmd) + result = 0 + else: + print("执行 数据导入命令:", cmd) + result = os.system(cmd) + if result == 0: + successSql.append(sql) + index = sqlid + else: + # 记录错误退出 + print("执行sql出错") + break + elif expectIndex > sqlid: # 出现了不连续就停止 + continue + else: + print("出现不文件id连续 下一个应该为：%d，实际为 %d" % (index+1, sqlid)) + break + # 记录日志 + if len(successSql)>0: + # 记录日志 + conf['fileId'] = str(index) + if debug: + print("设置配置 fileId:", index) + else: + print("保存 设置配置 fileId:", index) + with open(configFile, 'w') as c: + config.write(c) + + # 移除文件 + for sql in successSql: + cmd = "mv -f %s%s %s" % (conf['sqlFilePath'], sql, conf['sqlFileBakPath']) + if debug: + print("移除已执行文件命令:", cmd) + else: + print("执行 移除已执行文件命令:", cmd) + os.system(cmd) + else: + print("没有需要导入的文件") + + else: + print("没有需要处理的文件") +