diff options
Diffstat (limited to 'storage/rocksdb/myrocks_hotbackup.py')
| -rwxr-xr-x | storage/rocksdb/myrocks_hotbackup.py | 697 |
1 files changed, 697 insertions, 0 deletions
diff --git a/storage/rocksdb/myrocks_hotbackup.py b/storage/rocksdb/myrocks_hotbackup.py new file mode 100755 index 00000000..42c25c95 --- /dev/null +++ b/storage/rocksdb/myrocks_hotbackup.py @@ -0,0 +1,697 @@ +#!@PYTHON_SHEBANG@ + +from __future__ import division +from optparse import OptionParser +import collections +import signal +import os +import stat +import sys +import re +import subprocess +import logging +import logging.handlers +import time +import datetime +import shutil +import traceback +import tempfile + +import MySQLdb +import MySQLdb.connections +from MySQLdb import OperationalError, ProgrammingError + +logger = None +opts = None +rocksdb_files = ['MANIFEST', 'CURRENT', 'OPTIONS'] +rocksdb_data_suffix = '.sst' +rocksdb_wal_suffix = '.log' +exclude_files = ['master.info', 'relay-log.info', 'worker-relay-log.info', + 'auto.cnf', 'gaplock.log', 'ibdata', 'ib_logfile', '.trash'] +wdt_bin = 'wdt' + +def is_manifest(fname): + for m in rocksdb_files: + if fname.startswith(m): + return True + return False + +class Writer(object): + a = None + def __init__(self): + a = None + +class StreamWriter(Writer): + stream_cmd= '' + + def __init__(self, stream_option, direct = 0): + super(StreamWriter, self).__init__() + if stream_option == 'tar': + self.stream_cmd= 'tar chf -' + elif stream_option == 'xbstream': + self.stream_cmd= 'xbstream -c' + if direct: + self.stream_cmd = self.stream_cmd + ' -d' + else: + raise Exception("Only tar or xbstream is supported as streaming option.") + + def write(self, file_name): + rc= os.system(self.stream_cmd + " " + file_name) + if (rc != 0): + raise Exception("Got error on stream write: " + str(rc) + " " + file_name) + + +class MiscFilesProcessor(): + datadir = None + wildcard = r'.*\.[frm|MYD|MYI|MAD|MAI|MRG|TRG|TRN|ARM|ARZ|CSM|CSV|opt|par]' + regex = None + start_backup_time = None + skip_check_frm_timestamp = None + + def __init__(self, datadir, skip_check_frm_timestamp, start_backup_time): + self.datadir = datadir + self.regex = re.compile(self.wildcard) + self.skip_check_frm_timestamp = skip_check_frm_timestamp + self.start_backup_time = start_backup_time + + def process_db(self, db): + # do nothing + pass + + def process_file(self, path): + # do nothing + pass + + def check_frm_timestamp(self, fname, path): + if not self.skip_check_frm_timestamp and fname.endswith('.frm'): + if os.path.getmtime(path) > self.start_backup_time: + logger.error('FRM file %s was updated after starting backups. ' + 'Schema could have changed and the resulting copy may ' + 'not be valid. Aborting. ' + '(backup time: %s, file modifled time: %s)', + path, datetime.datetime.fromtimestamp(self.start_backup_time).strftime('%Y-%m-%d %H:%M:%S'), + datetime.datetime.fromtimestamp(os.path.getmtime(path)).strftime('%Y-%m-%d %H:%M:%S')) + raise Exception("Inconsistent frm file timestamp"); + + def process(self): + os.chdir(self.datadir) + for db in self.get_databases(): + logger.info("Starting MySQL misc file traversal from database %s..", db) + self.process_db(db) + for f in self.get_files(db): + if self.match(f): + rel_path = os.path.join(db, f) + self.check_frm_timestamp(f, rel_path) + self.process_file(rel_path) + logger.info("Traversing misc files from data directory..") + for f in self.get_files(""): + should_skip = False + for e in exclude_files: + if f.startswith(e) or f.endswith(e): + logger.info("Skipping %s", f) + should_skip = True + break + if not should_skip: + self.process_file(f) + + def match(self, filename): + if self.regex.match(filename): + return True + else: + return False + + def get_databases(self): + dbs = [] + dirs = [ d for d in os.listdir(self.datadir) \ + if not os.path.isfile(os.path.join(self.datadir,d))] + for db in dirs: + if not db.startswith('.') and not self._is_socket(db) and not db == "#rocksdb": + dbs.append(db) + return dbs + + def get_files(self, db): + dbdir = self.datadir + "/" + db + return [ f for f in os.listdir(dbdir) \ + if os.path.isfile(os.path.join(dbdir,f))] + + def _is_socket(self, item): + mode = os.stat(os.path.join(self.datadir, item)).st_mode + if stat.S_ISSOCK(mode): + return True + return False + + +class MySQLBackup(MiscFilesProcessor): + writer = None + + def __init__(self, datadir, writer, skip_check_frm_timestamp, start_backup_time): + MiscFilesProcessor.__init__(self, datadir, skip_check_frm_timestamp, start_backup_time) + self.writer = writer + + def process_file(self, fname): # overriding base class + self.writer.write(fname) + + +class MiscFilesLinkCreator(MiscFilesProcessor): + snapshot_dir = None + + def __init__(self, datadir, snapshot_dir, skip_check_frm_timestamp, start_backup_time): + MiscFilesProcessor.__init__(self, datadir, skip_check_frm_timestamp, start_backup_time) + self.snapshot_dir = snapshot_dir + + def process_db(self, db): + snapshot_sub_dir = os.path.join(self.snapshot_dir, db) + os.makedirs(snapshot_sub_dir) + + def process_file(self, path): + dst_path = os.path.join(self.snapshot_dir, path) + os.link(path, dst_path) + + +# RocksDB backup +class RocksDBBackup(): + source_dir = None + writer = None + # sst files sent in this backup round + sent_sst = {} + # target sst files in this backup round + target_sst = {} + # sst files sent in all backup rounds + total_sent_sst= {} + # sum of sst file size sent in this backup round + sent_sst_size = 0 + # sum of target sst file size in this backup round + # if sent_sst_size becomes equal to target_sst_size, + # it means the backup round finished backing up all sst files + target_sst_size = 0 + # sum of all sst file size sent all backup rounds + total_sent_sst_size= 0 + # sum of all target sst file size from all backup rounds + total_target_sst_size = 0 + show_progress_size_interval= 1073741824 # 1GB + wal_files= [] + manifest_files= [] + finished= False + + def __init__(self, source_dir, writer, prev): + self.source_dir = source_dir + self.writer = writer + os.chdir(self.source_dir) + self.init_target_files(prev) + + def init_target_files(self, prev): + sst = {} + self.sent_sst = {} + self.target_sst= {} + self.total_sent_sst = {} + self.sent_sst_size = 0 + self.target_sst_size = 0 + self.total_sent_sst_size= 0 + self.total_target_sst_size= 0 + self.wal_files= [] + self.manifest_files= [] + + for f in os.listdir(self.source_dir): + if f.endswith(rocksdb_data_suffix): + # exactly the same file (same size) was sent in previous backup rounds + if prev is not None and f in prev.total_sent_sst and int(os.stat(f).st_size) == prev.total_sent_sst[f]: + continue + sst[f]= int(os.stat(f).st_size) + self.target_sst_size = self.target_sst_size + os.stat(f).st_size + elif is_manifest(f): + self.manifest_files.append(f) + elif f.endswith(rocksdb_wal_suffix): + self.wal_files.append(f) + self.target_sst= collections.OrderedDict(sorted(sst.items())) + + if prev is not None: + self.total_sent_sst = prev.total_sent_sst + self.total_sent_sst_size = prev.total_sent_sst_size + self.total_target_sst_size = self.target_sst_size + prev.total_sent_sst_size + else: + self.total_target_sst_size = self.target_sst_size + + def do_backup_single(self, fname): + self.writer.write(fname) + os.remove(fname) + + def do_backup_sst(self, fname, size): + self.do_backup_single(fname) + self.sent_sst[fname]= size + self.total_sent_sst[fname]= size + self.sent_sst_size = self.sent_sst_size + size + self.total_sent_sst_size = self.total_sent_sst_size + size + + def do_backup_manifest(self): + for f in self.manifest_files: + self.do_backup_single(f) + + def do_backup_wal(self): + for f in self.wal_files: + self.do_backup_single(f) + + # this is the last snapshot round. backing up all the rest files + def do_backup_final(self): + logger.info("Backup WAL..") + self.do_backup_wal() + logger.info("Backup Manifest..") + self.do_backup_manifest() + self.do_cleanup() + self.finished= True + + def do_cleanup(self): + shutil.rmtree(self.source_dir) + logger.info("Cleaned up checkpoint from %s", self.source_dir) + + def do_backup_until(self, time_limit): + logger.info("Starting backup from snapshot: target files %d", len(self.target_sst)) + start_time= time.time() + last_progress_time= start_time + progress_size= 0 + for fname, size in self.target_sst.iteritems(): + self.do_backup_sst(fname, size) + progress_size= progress_size + size + elapsed_seconds = time.time() - start_time + progress_seconds = time.time() - last_progress_time + + if self.should_show_progress(size): + self.show_progress(progress_size, progress_seconds) + progress_size=0 + last_progress_time= time.time() + + if elapsed_seconds > time_limit and self.has_sent_all_sst() is False: + logger.info("Snapshot round finished. Elapsed Time: %5.2f. Remaining sst files: %d", + elapsed_seconds, len(self.target_sst) - len(self.sent_sst)) + self.do_cleanup() + break; + if self.has_sent_all_sst(): + self.do_backup_final() + + return self + + def should_show_progress(self, size): + if int(self.total_sent_sst_size/self.show_progress_size_interval) > int((self.total_sent_sst_size-size)/self.show_progress_size_interval): + return True + else: + return False + + def show_progress(self, size, seconds): + logger.info("Backup Progress: %5.2f%% Sent %6.2f GB of %6.2f GB data, Transfer Speed: %6.2f MB/s", + self.total_sent_sst_size*100/self.total_target_sst_size, + self.total_sent_sst_size/1024/1024/1024, + self.total_target_sst_size/1024/1024/1024, + size/seconds/1024/1024) + + def print_backup_report(self): + logger.info("Sent %6.2f GB of sst files, %d files in total.", + self.total_sent_sst_size/1024/1024/1024, + len(self.total_sent_sst)) + + def has_sent_all_sst(self): + if self.sent_sst_size == self.target_sst_size: + return True + return False + + +class MySQLUtil: + @staticmethod + def connect(user, password, port, socket=None): + if socket: + dbh = MySQLdb.Connect(user=user, + passwd=password, + unix_socket=socket) + else: + dbh = MySQLdb.Connect(user=user, + passwd=password, + port=port, + host="127.0.0.1") + return dbh + + @staticmethod + def create_checkpoint(dbh, checkpoint_dir): + sql = ("SET GLOBAL rocksdb_create_checkpoint='{0}'" + .format(checkpoint_dir)) + cur= dbh.cursor() + cur.execute(sql) + cur.close() + + @staticmethod + def get_datadir(dbh): + sql = "SELECT @@datadir" + cur = dbh.cursor() + cur.execute(sql) + row = cur.fetchone() + return row[0] + + @staticmethod + def is_directio_enabled(dbh): + sql = "SELECT @@global.rocksdb_use_direct_reads" + cur = dbh.cursor() + cur.execute(sql) + row = cur.fetchone() + return row[0] + +class BackupRunner: + datadir = None + start_backup_time = None + + def __init__(self, datadir): + self.datadir = datadir + self.start_backup_time = time.time() + + def start_backup_round(self, backup_round, prev_backup): + def signal_handler(*args): + logger.info("Got signal. Exit") + if b is not None: + logger.info("Cleaning up snapshot directory..") + b.do_cleanup() + sys.exit(1) + + b = None + try: + signal.signal(signal.SIGINT, signal_handler) + w = None + if not opts.output_stream: + raise Exception("Currently only streaming backup is supported.") + + snapshot_dir = opts.checkpoint_directory + '/' + str(backup_round) + dbh = MySQLUtil.connect(opts.mysql_user, + opts.mysql_password, + opts.mysql_port, + opts.mysql_socket) + direct = MySQLUtil.is_directio_enabled(dbh) + logger.info("Direct I/O: %d", direct) + + w = StreamWriter(opts.output_stream, direct) + + if not self.datadir: + self.datadir = MySQLUtil.get_datadir(dbh) + logger.info("Set datadir: %s", self.datadir) + logger.info("Creating checkpoint at %s", snapshot_dir) + MySQLUtil.create_checkpoint(dbh, snapshot_dir) + logger.info("Created checkpoint at %s", snapshot_dir) + b = RocksDBBackup(snapshot_dir, w, prev_backup) + return b.do_backup_until(opts.checkpoint_interval) + except Exception as e: + logger.error(e) + logger.error(traceback.format_exc()) + if b is not None: + logger.info("Cleaning up snapshot directory.") + b.do_cleanup() + sys.exit(1) + + def backup_mysql(self): + try: + w = None + if opts.output_stream: + w = StreamWriter(opts.output_stream) + else: + raise Exception("Currently only streaming backup is supported.") + b = MySQLBackup(self.datadir, w, opts.skip_check_frm_timestamp, + self.start_backup_time) + logger.info("Taking MySQL misc backups..") + b.process() + logger.info("MySQL misc backups done.") + except Exception as e: + logger.error(e) + logger.error(traceback.format_exc()) + sys.exit(1) + + +class WDTBackup: + datadir = None + start_backup_time = None + + def __init__(self, datadir): + self.datadir = datadir + self.start_backup_time = time.time() + + def cleanup(self, snapshot_dir, server_log): + if server_log: + server_log.seek(0) + logger.info("WDT server log:") + logger.info(server_log.read()) + server_log.close() + if snapshot_dir: + logger.info("Cleaning up snapshot dir %s", snapshot_dir) + shutil.rmtree(snapshot_dir) + + def backup_with_timeout(self, backup_round): + def signal_handler(*args): + logger.info("Got signal. Exit") + self.cleanup(snapshot_dir, server_log) + sys.exit(1) + + logger.info("Starting backup round %d", backup_round) + snapshot_dir = None + server_log = None + try: + signal.signal(signal.SIGINT, signal_handler) + # create rocksdb snapshot + snapshot_dir = os.path.join(opts.checkpoint_directory, str(backup_round)) + dbh = MySQLUtil.connect(opts.mysql_user, + opts.mysql_password, + opts.mysql_port, + opts.mysql_socket) + logger.info("Creating checkpoint at %s", snapshot_dir) + MySQLUtil.create_checkpoint(dbh, snapshot_dir) + logger.info("Created checkpoint at %s", snapshot_dir) + + # get datadir if not provided + if not self.datadir: + self.datadir = MySQLUtil.get_datadir(dbh) + logger.info("Set datadir: %s", self.datadir) + + # create links for misc files + link_creator = MiscFilesLinkCreator(self.datadir, snapshot_dir, + opts.skip_check_frm_timestamp, + self.start_backup_time) + link_creator.process() + + current_path = os.path.join(opts.backupdir, "CURRENT") + + # construct receiver cmd, using the data directory as recovery-id. + # we delete the current file because it is not append-only, therefore not + # resumable. + remote_cmd = ( + "ssh {0} rm -f {1}; " + "{2} -directory {3} -enable_download_resumption " + "-recovery_id {4} -start_port 0 -abort_after_seconds {5} {6}" + ).format(opts.destination, + current_path, + wdt_bin, + opts.backupdir, + self.datadir, + opts.checkpoint_interval, + opts.extra_wdt_receiver_options) + logger.info("WDT remote cmd %s", remote_cmd) + server_log = tempfile.TemporaryFile() + remote_process = subprocess.Popen(remote_cmd.split(), + stdout=subprocess.PIPE, + stderr=server_log) + wdt_url = remote_process.stdout.readline().strip() + if not wdt_url: + raise Exception("Unable to get connection url from wdt receiver") + sender_cmd = ( + "{0} -connection_url \'{1}\' -directory {2} -app_name=myrocks " + "-avg_mbytes_per_sec {3} " + "-enable_download_resumption -abort_after_seconds {4} {5}" + ).format(wdt_bin, + wdt_url, + snapshot_dir, + opts.avg_mbytes_per_sec, + opts.checkpoint_interval, + opts.extra_wdt_sender_options) + sender_status = os.system(sender_cmd) >> 8 + remote_status = remote_process.wait() + self.cleanup(snapshot_dir, server_log) + # TODO: handle retryable and non-retyable errors differently + return (sender_status == 0 and remote_status == 0) + + except Exception as e: + logger.error(e) + logger.error(traceback.format_exc()) + self.cleanup(snapshot_dir, server_log) + sys.exit(1) + + +def backup_using_wdt(): + if not opts.destination: + logger.error("Must provide remote destination when using WDT") + sys.exit(1) + + # TODO: detect whether WDT is installed + logger.info("Backing up myrocks to %s using WDT", opts.destination) + wdt_backup = WDTBackup(opts.datadir) + finished = False + backup_round = 1 + while not finished: + start_time = time.time() + finished = wdt_backup.backup_with_timeout(backup_round) + end_time = time.time() + duration_seconds = end_time - start_time + if (not finished) and (duration_seconds < opts.checkpoint_interval): + # round finished before timeout + sleep_duration = (opts.checkpoint_interval - duration_seconds) + logger.info("Sleeping for %f seconds", sleep_duration) + time.sleep(sleep_duration) + + backup_round = backup_round + 1 + logger.info("Finished myrocks backup using WDT") + + +def init_logger(): + global logger + logger = logging.getLogger('myrocks_hotbackup') + logger.setLevel(logging.INFO) + h1= logging.StreamHandler(sys.stderr) + f = logging.Formatter("%(asctime)s.%(msecs)03d %(levelname)s %(message)s", + "%Y-%m-%d %H:%M:%S") + h1.setFormatter(f) + logger.addHandler(h1) + +backup_wdt_usage = ("Backup using WDT: myrocks_hotbackup " + "--user=root --password=pw --stream=wdt " + "--checkpoint_dir=<directory where temporary backup hard links " + "are created> --destination=<remote host name> --backup_dir=" + "<remote directory name>. This has to be executed at the src " + "host.") +backup_usage= "Backup: set -o pipefail; myrocks_hotbackup --user=root --password=pw --port=3306 --checkpoint_dir=<directory where temporary backup hard links are created> | ssh -o NoneEnabled=yes remote_server 'tar -xi -C <directory on remote server where backups will be sent>' . You need to execute backup command on a server where you take backups." +move_back_usage= "Move-Back: myrocks_hotbackup --move_back --datadir=<dest mysql datadir> --rocksdb_datadir=<dest rocksdb datadir> --rocksdb_waldir=<dest rocksdb wal dir> --backup_dir=<where backup files are stored> . You need to execute move-back command on a server where backup files are sent." + + +def parse_options(): + global opts + parser = OptionParser(usage = "\n\n" + backup_usage + "\n\n" + \ + backup_wdt_usage + "\n\n" + move_back_usage) + parser.add_option('-i', '--interval', type='int', dest='checkpoint_interval', + default=300, + help='Number of seconds to renew checkpoint') + parser.add_option('-c', '--checkpoint_dir', type='string', dest='checkpoint_directory', + default='/data/mysql/backup/snapshot', + help='Local directory name where checkpoints will be created.') + parser.add_option('-d', '--datadir', type='string', dest='datadir', + default=None, + help='backup mode: src MySQL datadir. move_back mode: dest MySQL datadir') + parser.add_option('-s', '--stream', type='string', dest='output_stream', + default='tar', + help='Setting streaming backup options. Currently tar, WDT ' + 'and xbstream are supported. Default is tar') + parser.add_option('--destination', type='string', dest='destination', + default='', + help='Remote server name. Only used for WDT mode so far.') + parser.add_option('--avg_mbytes_per_sec', type='int', + dest='avg_mbytes_per_sec', + default=500, + help='Average backup rate in MBytes/sec. WDT only.') + parser.add_option('--extra_wdt_sender_options', type='string', + dest='extra_wdt_sender_options', + default='', + help='Extra options for WDT sender') + parser.add_option('--extra_wdt_receiver_options', type='string', + dest='extra_wdt_receiver_options', + default='', + help='Extra options for WDT receiver') + parser.add_option('-u', '--user', type='string', dest='mysql_user', + default='root', + help='MySQL user name') + parser.add_option('-p', '--password', type='string', dest='mysql_password', + default='', + help='MySQL password name') + parser.add_option('-P', '--port', type='int', dest='mysql_port', + default=3306, + help='MySQL port number') + parser.add_option('-S', '--socket', type='string', dest='mysql_socket', + default=None, + help='MySQL socket path. Takes precedence over --port.') + parser.add_option('-m', '--move_back', action='store_true', dest='move_back', + default=False, + help='Moving MyRocks backup files to proper locations.') + parser.add_option('-r', '--rocksdb_datadir', type='string', dest='rocksdb_datadir', + default=None, + help='RocksDB target data directory where backup data files will be moved. Must be empty.') + parser.add_option('-w', '--rocksdb_waldir', type='string', dest='rocksdb_waldir', + default=None, + help='RocksDB target data directory where backup wal files will be moved. Must be empty.') + parser.add_option('-b', '--backup_dir', type='string', dest='backupdir', + default=None, + help='backup mode for WDT: Remote directory to store ' + 'backup. move_back mode: Locations where backup ' + 'files are stored.') + parser.add_option('-f', '--skip_check_frm_timestamp', + dest='skip_check_frm_timestamp', + action='store_true', default=False, + help='skipping to check if frm files are updated after starting backup.') + parser.add_option('-D', '--debug_signal_file', type='string', dest='debug_signal_file', + default=None, + help='debugging purpose: waiting until the specified file is created') + + opts, args = parser.parse_args() + + +def create_moveback_dir(directory): + if not os.path.exists(directory): + os.makedirs(directory) + else: + for f in os.listdir(directory): + logger.error("Directory %s has file or directory %s!", directory, f) + raise + +def print_move_back_usage(): + logger.warning(move_back_usage) + +def move_back(): + if opts.rocksdb_datadir is None or opts.rocksdb_waldir is None or opts.backupdir is None or opts.datadir is None: + print_move_back_usage() + sys.exit() + create_moveback_dir(opts.datadir) + create_moveback_dir(opts.rocksdb_datadir) + create_moveback_dir(opts.rocksdb_waldir) + + os.chdir(opts.backupdir) + for f in os.listdir(opts.backupdir): + if os.path.isfile(os.path.join(opts.backupdir,f)): + if f.endswith(rocksdb_wal_suffix): + shutil.move(f, opts.rocksdb_waldir) + elif f.endswith(rocksdb_data_suffix) or is_manifest(f): + shutil.move(f, opts.rocksdb_datadir) + else: + shutil.move(f, opts.datadir) + else: #directory + if f.endswith('.rocksdb'): + continue + shutil.move(f, opts.datadir) + +def start_backup(): + logger.info("Starting backup.") + runner = BackupRunner(opts.datadir) + b = None + backup_round= 1 + while True: + b = runner.start_backup_round(backup_round, b) + backup_round = backup_round + 1 + if b.finished is True: + b.print_backup_report() + logger.info("RocksDB Backup Done.") + break + if opts.debug_signal_file: + while not os.path.exists(opts.debug_signal_file): + logger.info("Waiting until %s is created..", opts.debug_signal_file) + time.sleep(1) + runner.backup_mysql() + logger.info("All Backups Done.") + + +def main(): + parse_options() + init_logger() + + if opts.move_back is True: + move_back() + elif opts.output_stream == 'wdt': + backup_using_wdt() + else: + start_backup() + +if __name__ == "__main__": + main() |