path: root/storage/rocksdb/myrocks_hotbackup.py

#!@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()