host unit backup script by lvm thinprovisioning and snapshot

lvm-host-backup.py

#!/usr/bin/env python3
# vim:fileencoding=utf-8
from datetime import datetime
from argparse import ArgumentParser, FileType
from concurrent.futures import ThreadPoolExecutor, as_completed
from subprocess import Popen, PIPE
import threading
import re
import shlex
import time
import logging
import logging.handlers
import importlib.machinery
import os, sys, io

__author__  = 'holly'
__version__ = '1.0'

DESCRIPTION  = 'lvm backup cli'
CONFIG_DIR   = '/etc/lvm-host-backup'
RSYNC_CONFIG = os.path.join(CONFIG_DIR, 'rsync_config.py')
RSYNC_TARGET = os.path.join(CONFIG_DIR, 'rsync_target.py')

###########################################
#
# * sample rsync_config.py 
#
# ```
# remote_rsync_path = 'sudo ionice -c3 nice -n 19 /usr/bin/rsync'
# ssh_user = 'backup'
# ssh_privkey  = '/root/.ssh/backup_id_rsa'
# ```
#
# * sample rsync_target.py 
#
# ```
# target_hosts = [
#        {'src_host': '172.16.0.4', 'target_dirs': ['/etc', '/opt', '/home']},
#        {'src_host': '172.16.0.5', 'target_dirs': ['/etc', '/opt', '/home']},
#        {'src_host': '172.16.0.6', 'target_dirs': ['/var'], 'rsync_config': {"ssh_port": "9022"}},
#    ]
#
# ```
#
###########################################

LOG_FORMAT      = "%(asctime)s %(name)s %(levelname)s: %(message)s"
FORK            = 1
SNAPSHOT_ROTATE = 3

class Log():

    def __init__(self, log_file=None, quiet=False, debug=False):
        self._logger = logging.getLogger(__name__)

        self._logger.addHandler(logging.StreamHandler())
        if log_file:
            #fh = logging.FileHandler(log_file)
            fh = logging.handlers.TimedRotatingFileHandler(log_file, when="D", backupCount=30)
            fh.formatter = logging.Formatter(fmt=LOG_FORMAT)
            self._logger.addHandler(fh)

        if quiet:
            self._logger.setLevel(logging.CRITICAL)
        elif debug:
            self._logger.setLevel(logging.DEBUG)
        else:
            self._logger.setLevel(logging.INFO)

    def shutdown(self):
        logging.shutdown()

    @property
    def logger(self):
        return self._logger


class Rsync: 

    config = {}
    config['local_rsync_path']       = '/usr/bin/rsync'
    config['local_backup_dir']       = '/backup'
    config['local_config_dir']       = os.path.join(config['local_backup_dir'], ".config")
    config['local_snapshot_dir']     = os.path.join(config['local_backup_dir'], ".snapshot")
    config['remote_rsync_path']      = 'ionice -c3 nice -n 19 rsync'
    config['ssh_opts']               = ['ssh', '-2', '-x', '-T', '-o StrictHostKeyChecking=no', '-o UserKnownHostsFile=/dev/null', '-o Compression=no']
    config['ssh_privkey']            = '/root/.ssh/id_rsa'
    config['ssh_user']               = 'root'
    config['ssh_port']               = 22
    config['ssh_connection_timeout'] = 30
    config['ssh_cipher']             = 'aes256-ctr'
    config['rsync_bwlimit']          = 0
    config['rsync_opts']             = ['-axSHAX', '--numeric-ids', '--delete', '--timeout=0']
    config['rsync_other_opts']       = ['--append', '--partial']
    config['rsync_excludes']         = ['*.swp', '*.tmp', '*~', '.make.state', '.nse_depinfo', '#*', '.#*', ',*', '_$*', '*$', '*.old', '*.bak','*.BAK', '*.orig', '*.rej', '.del-*', '*.olb', '*.obj', '*.Z', '*.elc', '*.ln', 'core']
    config['rsync_exclude_from']     = os.path.join(config['local_config_dir'], "{0}", "exclude_from")
    config['rsync_sp_opts_file']     = os.path.join(config['local_config_dir'], "{0}", "sp_opts")

    config['lvm_vg_name']            = 'vg01'
    config['lvm_thin_name']          = '{0}/thinpool'.format(config['lvm_vg_name'])
    config['lvm_lv_vsize']           = '100GB'

    def __init__(self, src_host, rsync_config):

        self.src_host    = src_host
        self.date_string = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.config = self.parse_config(rsync_config)

    def parse_config(self, rsync_config):
        config = importlib.machinery.SourceFileLoader('config', rsync_config).load_module() if os.path.exists(rsync_config) else type('config', (object,), {})()

        for key in self.__class__.config:
            if not hasattr(config, key):
                setattr(config, key, self.__class__.config[key])
        return config

    def lvm_snapshots(self):
        snapshots = []
        # 192.168.0.4_20170312 vg01 thin,sparse public,snapshot,thinsnapshot 2017-03-12 10:20:06 +0900
        cmds = self.make_lvdisplay_commands()
        outs, errs, returncode = self._execute_command(cmds)
        if returncode != 0:
            raise Exception(errs)
        for line in re.split(r"\n", outs):
            line = line.strip()
            fields = re.split(r"\s+", line)
            if re.search(r"^{0}_\d{{8}}_\d{{6}}$".format(self.src_host), fields[0]) and fields[1] == self.config.lvm_vg_name and re.search(r"thinsnapshot", fields[3]):
                snapshots.append(fields[0])
        return snapshots

    def check_lvvol(self):
        cmds = self.make_lvdisplay_commands()
        outs, errs, returncode = self._execute_command(cmds)
        if returncode != 0:
            raise Exception(errs)

        for line in re.split("\n", outs):
            fields = re.split("\s+", line.strip())
            if fields[0] == self.src_host and fields[1] == self.config.lvm_vg_name:
                return True
        return False

    def check_mount(self):
        mountpoint = os.path.join(self.config.local_backup_dir, self.src_host)
        outs, errs, returncode = self._execute_command(["mountpoint", mountpoint])
        return True if returncode == 0 else False

    def check_snapshot_mount(self, snapshot_name):
        mountpoint = os.path.join(self.config.local_snapshot_dir, self.src_host, snapshot_name)
        outs, errs, returncode = self._execute_command(["mountpoint", mountpoint])
        return True if returncode == 0 else False

    def make_lvdisplay_commands(self):
        # LV                   VG   Layout      Role                         Time
        # lvol0                vg01 linear      public                       2017-03-05 23:58:26 +0900
        # thinpool             vg01 thin,pool   private                      2017-03-06 00:00:30 +0900
        # 192.168.0.4          vg01 thin,sparse public,origin,thinorigin     2017-03-12 02:49:58 +0900
        # 192.168.0.254        vg01 thin,sparse public                       2017-03-12 02:49:58 +0900
        # 192.168.0.4_20170312 vg01 thin,sparse public,snapshot,thinsnapshot 2017-03-12 10:20:06 +0900
        cmds = ["lvdisplay", "--columns", "--noheadings", "--sort", "lv_time", "--options", "lv_name,vg_name,lv_layout,lv_role,lv_time"]
        return cmds

    def make_lvdisplay_command_string(self):
        cmds = self.make_lvdisplay_commands()
        return " ".join(cmds)

    def make_lvcreate_commands(self):
        cmds = ["lvcreate", "-T", "-V", self.config.lvm_lv_vsize, "-n", self.src_host, self.config.lvm_thin_name]
        return cmds

    def make_lvcreate_command_string(self):
        cmds = self.make_lvcreate_commands()
        return " ".join(cmds)

    def make_snapshot_commands(self, snapshot_name=None):
        if not snapshot_name:
            snapshot_name = "{0}_{1}".format(self.src_host, self.date_string)
        lv_fullname   = "{0}/{1}".format(self.config.lvm_vg_name, self.src_host)
        cmds = ["lvcreate", "-s", "-n", snapshot_name, lv_fullname]
        return cmds

    def make_snapshot_command_string(self, snapshot_name=None):
        cmds = self.make_snapshot_commands(snapshot_name)
        return " ".join(cmds)

    def make_vgchange_commands(self):
        cmds = ["vgchange", "-ay", self.config.lvm_vg_name]
        return cmds

    def make_vgchange_command_string(self):
        cmds = self.make_vgchange_commands()
        return " ".join(cmds)

    def make_lvchange_commands(self, lv_name):
        lv_fullname = self.make_lv_fullname(lv_name)
        cmds = ["lvchange", "-kn", "-ay", lv_fullname]
        return cmds

    def make_lvchange_command_string(self, lv_name):
        cmds = self.make_lvchange_commands(lv_name)
        return " ".join(cmds)


    def make_lvremove_commands(self, lv_name):
        lv_fullname = "/dev/{0}/{1}".format(self.config.lvm_vg_name, lv_name)
        cmds = ["lvremove", "-y", lv_fullname]
        return cmds

    def make_lvremove_command_string(self, lv_name):
        cmds = self.make_lvremove_commands(lv_name)
        return " ".join(cmds)

    def make_fs_commands(self):
        device  = self.make_lv_fullname()
        cmds    = ["mkfs.xfs", "-f", "-b size=4096", "-i size=512", "-l size=64m", device]
        return cmds

    def make_fs_command_string(self):
        cmds = self.make_fs_commands()
        return " ".join(cmds)

    def make_mount_commands(self):
        cmds = [ "mount", "-t", "xfs" ]
        mountpoint  = os.path.join(self.config.local_backup_dir, self.src_host)
        device      = self.make_lv_fullname()
        mount_opts  = ["noatime", "nobarrier"]
        cmds.extend(["-o", ",".join(mount_opts)])
        cmds.extend([device, mountpoint])
        return cmds

    def make_mount_command_string(self):
        cmds = self.make_mount_commands()
        return " ".join(cmds)

    def make_mount_snapshot_commands(self, snapshot_name=None):
        cmds = [ "mount", "-t", "xfs" ]
        if snapshot_name is None:
            snapshot_name = "{0}_{1}".format(self.src_host, self.date_string)
        mountpoint  = os.path.join(self.config.local_snapshot_dir, self.src_host, snapshot_name)
        device      = self.make_lv_fullname(snapshot_name)
        mount_opts  = ["ro", "nouuid", "noatime"]
        cmds.extend(["-o", ",".join(mount_opts)])
        cmds.extend([device, mountpoint])
        return cmds

    def make_mount_snapshot_command_string(self, snapshot_name=None):
        cmds = self.make_mount_snapshot_commands(snapshot_name=snapshot_name)
        return " ".join(cmds)

    def make_umount_snapshot_commands(self, snapshot_name=None):
        mountpoint  = ""
        cmds = [ "umount", "-l", "-f" ]
        if snapshot_name is None:
            snapshot_name = "{0}_{1}".format(self.src_host, self.date_string)
        mountpoint  = os.path.join(self.config.local_snapshot_dir, self.src_host, snapshot_name)
        cmds.append(mountpoint)
        return cmds

    def make_umount_snapshot_command_string(self, snapshot_name=None):
        cmds = self.make_umount_snapshot_commands(snapshot_name=snapshot_name)
        return " ".join(cmds)

    def make_backup_host_dir(self):
        backup_host_dir = os.path.join(self.config.local_backup_dir, self.src_host)
        if not os.path.exists(backup_host_dir):
            os.makedirs(backup_host_dir, exist_ok=True)
        return backup_host_dir

    def make_snapshot_host_dir(self):
        snapshot_host_dir = os.path.join(self.config.local_snapshot_dir, self.src_host)
        if not os.path.exists(snapshot_host_dir):
            os.makedirs(snapshot_host_dir, exist_ok=True)
        return snapshot_host_dir

    def make_rsync_commands(self, src_dir, dry_run=False):

        config = self.config
        make_dir = True

        if dry_run:
            make_dir = False

        src = self.make_src_dir(src_dir)
        dst = self.make_dst_dir(src_dir, make_dir=make_dir)

        exclude_from = config.rsync_exclude_from.format(self.src_host)
        sp_opts_file = config.rsync_sp_opts_file.format(self.src_host)

        cmds   = [ config.local_rsync_path ]

        if os.path.exists(sp_opts_file):

            with open(sp_opts_file) as f:
                data = f.read().strip()
                data = re.sub(r"\\\n", " ", data)
                cmds.extend(shlex.split(data))
        else:
            cmds.extend(["--bwlimit={0}".format(config.rsync_bwlimit)])
            cmds.extend(config.rsync_opts)
            cmds.extend(config.rsync_other_opts)

            # shallow copy
            ssh_opts = list(config.ssh_opts)
            ssh_opts.append("-o ConnectTimeout={0}".format(config.ssh_connection_timeout))
            ssh_opts.append("-c {0}".format(config.ssh_cipher))
            ssh_opts.append("-p {0}".format(config.ssh_port))
            ssh_opts.append("-i {0}".format(config.ssh_privkey))
            ssh_opts.append("-l {0}".format(config.ssh_user))
            cmds.append("-e '{0}'".format(" ".join(ssh_opts)))
            cmds.append("--rsync-path='{0}'".format(config.remote_rsync_path))

            if os.path.exists(exclude_from):
                cmds.append("--exclude-from='{0}'".format(exclude_from))

            for exclude in config.rsync_excludes:
                cmds.append("--exclude='{0}'".format(exclude))

            if dry_run:
                cmds.append("--dry-run")

        cmds.append(src)
        cmds.append(dst)

        return cmds

    def make_rsync_command_string(self, src_dir, dry_run=False):
        cmds = self.make_rsync_commands(src_dir, dry_run=dry_run)
        return " ".join(cmds)

    def make_lv_fullname(self, lv_name=None):
        if lv_name is None:
            lv_name = self.src_host
        return "/dev/{0}/{1}".format(self.config.lvm_vg_name, lv_name)

    def make_src_dir(self, src_dir):
        m = re.search(r"^(.*)/+$", src_dir)
        if not m:
            src_dir = src_dir + "/"
        return "{0}:{1}".format(self.src_host, src_dir)

    def make_dst_dir(self, src_dir, make_dir=False):
        m = re.search(r"^(/+)(.*)$", src_dir)
        if m:
            src_dir = m.group(2)
        dst_dir = os.path.join(self.make_backup_host_dir(), src_dir)

        if make_dir:
            os.makedirs(dst_dir, exist_ok=True)

        return dst_dir

    def execute_rsync(self, src_dir, dry_run=False):
        cmds = self.make_rsync_commands(src_dir, dry_run=dry_run)
        return self._execute_command(cmds)

    def execute_from_string(self, command):
        return self._execute_command(shlex.split(command))

    def _execute_command(self, cmds):
        proc = Popen(cmds, universal_newlines=True, stdout=PIPE, stderr=PIPE)
        outs, errs = proc.communicate()
        return outs.strip(), errs.strip(), proc.returncode

parser = ArgumentParser(description=DESCRIPTION)
parser.add_argument('--version', action='version', version='%(prog)s ' + __version__)
parser.add_argument('--rsync-config', action='store', default=RSYNC_CONFIG, help='config file (Default: {0})'.format(RSYNC_CONFIG))
parser.add_argument('--rsync-target', action='store', default=RSYNC_TARGET, help='rsync target hosts file (Default: {0})'.format(RSYNC_TARGET))
parser.add_argument('--fork', '-f', action='store', type=int, default=FORK, help='backup fork processes (Default: {0})'.format(FORK))
parser.add_argument('--snapshot-rotate' ,'-s', action='store', type=int, default=SNAPSHOT_ROTATE, help='backup snapshot rotage age (Default: {0})'.format(SNAPSHOT_ROTATE))
parser.add_argument('--dry-run', '-n', action='store_true', default=False, help='backup rsync dry-run mode (Default: False)')
parser.add_argument('--log-file', '-l', action='store', help='output log file(Default: STDOUT)')
parser.add_argument('--mount-only', '-M', action='store_true', default=False, help='only lvm mount mode (Dfault: False)')
parser.add_argument('--quiet', '-q', action='store_true', default=False, help='quiet mode')
parser.add_argument('--verbose', action='store_true', help='output verbose message')
args = parser.parse_args()

log          = Log(log_file=args.log_file, quiet=args.quiet, debug=args.verbose)
rsync_target = importlib.machinery.SourceFileLoader('rsync_target', args.rsync_target).load_module()


def execute_command(rsync, command_string):
    command = re.split("\s+", command_string)[0]
    log.logger.info("target:{0} command:{1}".format(rsync.src_host, command_string))
    outs, errs, returncode = rsync.execute_from_string(command_string)
    if returncode == 0:
        log.logger.info("target:{0}: {1} success returncode:{2}".format(rsync.src_host, command, returncode))
    else:
        log.logger.error("target:{0} {1} failure returncode:{2}".format(rsync.src_host, command, returncode))
        raise Exception(errs)

def execute_rsync(target):

    src_host     = target["src_host"]
    target_dirs  = target["target_dirs"]
    rsync        = Rsync(src_host, args.rsync_config)

    thread_id    = threading.get_ident()

    start_time = int(datetime.now().strftime('%s'))
    log.logger.info(">> {0} backup start".format(src_host))
    log.logger.debug("thread_id {0}: start thread.".format(thread_id))

    backup_host_dir   = rsync.make_backup_host_dir()
    snapshot_host_dir = rsync.make_snapshot_host_dir()

    if rsync.check_lvvol():
        log.logger.info("lvvol {0} is exists".format(rsync.src_host))

        if args.mount_only or args.dry_run:
            log.logger.info("skip make and remove snapshot")
        else:
            # make new snapshot
            snapshot_name = "{0}_{1}".format(rsync.src_host, rsync.date_string)
            log.logger.info("make snapshot:{0}".format(snapshot_name))

            command_string = rsync.make_snapshot_command_string(snapshot_name)
            execute_command(rsync, command_string)

            # activate new snapshot
            command_string = rsync.make_lvchange_command_string(snapshot_name)
            execute_command(rsync, command_string)

            # delete old snapshot
            snapshots= rsync.lvm_snapshots()
            while args.snapshot_rotate < len(snapshots):
                command_strings = []
                if rsync.check_snapshot_mount(snapshots[0]):
                    command_strings.append(rsync.make_umount_snapshot_command_string(snapshots[0]))

                command_strings.append(rsync.make_lvremove_command_string(snapshots[0]))
                for command_string in command_strings:
                    execute_command(rsync, command_string)

                old_snapshot_dir = os.path.join(snapshot_host_dir, snapshots[0])
                os.removedirs(old_snapshot_dir)
                snapshots= rsync.lvm_snapshots()

    else:
        log.logger.info("make lvvol:{0} and filesystem".format(src_host))
        command_strings = [ rsync.make_lvcreate_command_string() ]
        command_strings.append(rsync.make_fs_command_string())
        for command_string in command_strings:
            execute_command(rsync, command_string)

    if rsync.check_mount():
        log.logger.info("{0}.mountpoint is exists".format(src_host))
    else:
        # activate new snapshot
        command_strings = [ rsync.make_vgchange_command_string() ]
        command_strings.append(rsync.make_mount_command_string())
        for command_string in command_strings:
            execute_command(rsync, command_string)

    # check and mount all snapshots
    for snapshot in rsync.lvm_snapshots():
        snapshot_dir = os.path.join(snapshot_host_dir, snapshot)
        os.makedirs(snapshot_dir, exist_ok=True)
        if not rsync.check_snapshot_mount(snapshot):
            command_string = rsync.make_mount_snapshot_command_string(snapshot)
            execute_command(rsync, command_string)

    if args.mount_only:
        return

    if "rsync_config" in target:
        for key in target["rsync_config"]:
            setattr(rsync.config, key, target["rsync_config"][key])

    for target_dir in target_dirs:

        command_string = rsync.make_rsync_command_string(target_dir, dry_run=args.dry_run)

        start_rsync_time = int(datetime.now().strftime('%s'))
        log.logger.info(">>>> target:{0}:{1} rsync start. command:{2}".format(src_host, target_dir, command_string))

        outs, errs, returncode = rsync.execute_from_string(command_string)
        end_rsync_time = int(datetime.now().strftime('%s'))

        if returncode == 0 or returncode == 24:
            log.logger.info(">>>> target:{0}:{1} rsync is finished successfully. returncode:{2} ({3} sec)".format(src_host, target_dir, returncode, (end_rsync_time - start_rsync_time)))
        else:
            log.logger.error(">>>> target:{0}:{1} rsync is terminated abnormally. returncode:{2} ({3} sec)".format(src_host, target_dir, returncode, (end_rsync_time - start_rsync_time)))
            log.logger.error(errs)

    end_time = int(datetime.now().strftime('%s'))
    log.logger.info(">> {0} backup is finished ({1} sec)".format(src_host, (end_time - start_time)))
    log.logger.debug("thread_id {0}: break thread.".format(thread_id))

def main():
    """ [FUNCTIONS] method or function description
    """
    start_time = int(datetime.now().strftime('%s'))
    log.logger.info("===== start lvm host backup =====")

    with ThreadPoolExecutor(max_workers=args.fork) as executor:
        future_to_targets = { executor.submit(execute_rsync, target): target for target in rsync_target.target_hosts }
        for future in as_completed(future_to_targets):
            target = future_to_targets[future]
            try:
                data = future.result()
            except Exception as exc:
                log.logger.critical('%s generated an exception: %s' % (target, exc))
            else:
                pass

    end_time = int(datetime.now().strftime('%s'))
    log.logger.info("===== end lvm host backup ({0} sec) =====".format(end_time - start_time))
    sys.exit(0)

if __name__ == "__main__":
    main()