qfs_backup

#!/usr/bin/env python
#
# Author: Christopher Zimmerman
#
# Copyright 2012,2016 Quantcast Corporation. All rights reserved.
#
# This file is part of Quantcast File System (QFS).
#
# Licensed under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the License for the specific language governing
# permissions and limitations under the License.
#
# This tool can be used to generate a tar archive stream of the latest
# checkpoint and associated transaction logs to standard output.

from __future__ import print_function

import os
import re
import subprocess
import sys
from optparse import IndentedHelpFormatter, OptionParser

PRUNE_MSG = """
The log, and checkpoint pruning related options are for backward compatibility
only. These options have no effect, as pruning no longer performed by this
script. The meta server maintains log and checkpoint directories, and performs
checkpoint, and log pruning. The number of checkpoints and log segments to keep
and related parameters are specified by the meta server configuration.
"""


def check_directory(e, name, dir):
    if os.path.exists(dir):
        if not os.path.isdir(dir):
            e.append("'%s' is not a directory: %s" % (name, dir))
    else:
        e.append("'%s' directory does not exist: %s" % (name, dir))


def check_file(e, name, file):
    if os.path.exists(file):
        if not os.path.isfile(file):
            e.append("'%s' is not a file: %s" % (name, file))
    else:
        e.append("'%s' file does not exist: %s" % (name, file))


def parse_options():
    formatter = IndentedHelpFormatter(max_help_position=50, width=80)
    usage = "usage: ./%prog -c CHECKPOINT -t TRANSACTIONS"
    desc = """Generates a tar archive stream of the latest
checkpoint and associated transaction logs to standard output."""
    parser = OptionParser(
        usage, formatter=formatter, add_help_option=False, description=desc
    )

    parser.add_option(
        "-c",
        "--checkpoint",
        metavar="DIR",
        help="Path to checkpoint directory.",
    )
    parser.add_option(
        "-t",
        "--transactions",
        metavar="DIR",
        help="Path to transaction log directory.",
    )
    parser.add_option(
        "-o",
        "--count",
        default=4,
        type="int",
        help="Number of checkpoints to archive.",
    )
    parser.add_option(
        "-n",
        "--number",
        default=8,
        type="int",
        help="""Number of checkpoints to keep, overrides -p
-- Deprecated for the same reason as -p""",
    )
    parser.add_option(
        "-a",
        "--tar",
        default="/bin/tar",
        metavar="BINARY",
        help="Path to tar binary.",
    )
    parser.add_option(
        "-p",
        "--prune",
        action="store_true",
        help="Prune checkpoints -- Deprecated." + PRUNE_MSG,
    )
    parser.add_option(
        "-d", "--dry-run", action="store_true", help="Run in dry run mode."
    )
    parser.add_option(
        "-h",
        "--help",
        action="store_true",
        help="Print this help message and exit.",
    )

    for option in parser.option_list:
        if option.default != ("NO", "DEFAULT") and option.default != "":
            option.help += " (%default)"

    (options, args) = parser.parse_args()

    if options.help:
        parser.print_help()
        print()
        sys.exit(0)

    e = []
    if options.checkpoint:
        check_directory(e, "checkpoint", options.checkpoint)
    else:
        e.append("'checkpoint' directory must be specified")

    if options.transactions:
        check_directory(e, "transactions", options.transactions)
    else:
        e.append("'transactions' directory must be specified")

    if options.count < 1:
        e.append("'count' must be >= 1")
    else:
        options.count = int(options.count)

    if options.number < 1:
        e.append("'number' must be >= 1")
    else:
        options.number = int(options.number)

    if len(e) > 0:
        parser.print_help()
        print()
        for error in e:
            print("*** %s" % error)
        print()
        sys.exit(1)

    return options


def get_checkpoints(checkpoint_dir):
    """Return a list of checkpoint files in age ascending order."""

    checkpoints = []
    for file in os.listdir(checkpoint_dir):
        if file.startswith("chkpt"):
            checkpoint_number = find_checkpoint_number(file)
            if checkpoint_number:
                checkpoints.append(os.path.join(checkpoint_dir, file))

    checkpoints.sort(key=find_checkpoint_number, reverse=True)

    log("Found %s checkpoint files." % len(checkpoints))

    return checkpoints


def get_transactions(transactions_dir):
    """Return a list of transaction logs in age ascending order."""

    transactions = []
    for file in os.listdir(transactions_dir):
        if file.startswith("log"):
            transactions.append(os.path.join(transactions_dir, file))

    transactions.sort(key=find_transaction_number, reverse=True)

    log("Found %s transactions logs." % len(transactions))

    return transactions


def check_transaction_sequences(transactions):
    log("Checking transaction log sequences.")
    last_seq = None
    last_log_file = None
    for log_file in transactions:
        seq = find_transaction_log_segment_number(log_file)
        if last_seq and seq and last_seq - seq > 1:
            log(
                "WARNING missing transaction log between %s and %s"
                % (last_log_file, log_file)
            )
        last_log_file = log_file
        last_seq = seq


def find_transaction_log_segment_number(transaction):
    """Extract transaction log segment number from the file name."""

    this_number = None
    match = re.match(r"^.*?log\.\d+\.\d+\.\d+\.(\d+)$", transaction)
    if match:
        this_number = int(match.group(1))

    return this_number


def get_log_sequence(re_prefix, re_suffix, name):
    """Extract log sequence number from name, and convert it into
    string that can sorted in lexicographic order"""

    this_number = None
    match = re.match(re_prefix + r"\.(\d+)\.(\d+)\.(\d+)" + re_suffix, name)
    if match:
        this_number = ""
        for i in range(1, 4):
            this_number += "%016x" % int(match.group(i))

    return this_number


def find_checkpoint_number(checkpoint):
    """Extract checkpoint log sequence number from the file name."""

    return get_log_sequence("^.*?chkpt", "$", checkpoint)


def find_transaction_number(transaction):
    """Extract transaction log sequence number from the file name."""

    return get_log_sequence(r"^.*?log", r"\.\d+$", transaction)


def find_checkpoint_log_number(checkpoint):
    """Pull the transaction log sequence number referenced by a checkpoint
    file."""

    log_number = 0
    file = open(checkpoint, "r")
    for line in file:
        if not line.startswith("log/"):
            continue
        log_name = line[4:]
        log_number = find_transaction_number(log_name)
        break
    file.close()

    return [log_name, log_number]


def find_checkpoint_transactions(checkpoint, transactions):
    """Find the list of transaction logs needed by the specified
    checkpoint."""

    checkpoint_transactions = []
    [log_name, log_number] = find_checkpoint_log_number(checkpoint)

    log("Oldest log to archive: %s" % log_name)

    for logfile in transactions:
        this_number = find_transaction_number(logfile)
        if this_number:
            if this_number >= log_number:
                checkpoint_transactions.append(logfile)
                os.stat(logfile)

    return checkpoint_transactions


def get_latest_checkpoint(checkpoint_dir):
    return os.path.join(checkpoint_dir, "latest")


def get_last_transaction_log(transaction_dir):
    return os.path.join(transaction_dir, "last")


def log(msg):
    print(msg, file=sys.stderr)


if __name__ == "__main__":
    o = parse_options()

    log("Scanning: %s and %s" % (o.checkpoint, o.transactions))
    log("Archiving %s checkpoints at most." % o.count)
    if o.prune:
        log(PRUNE_MSG)

    checkpoints = get_checkpoints(o.checkpoint)

    if 0 == len(checkpoints):
        log("No checkpoints found exiting ...")
        sys.exit(1)

    tar_cmd = [o.tar, "-cf", "-"]

    filelist = []
    archive_checkpoints = []
    oldest_checkpoint = None
    if o.count <= len(checkpoints):
        oldest_checkpoint = checkpoints[o.count - 1]
        archive_checkpoints = checkpoints[: o.count]
    else:
        oldest_checkpoint = checkpoints[-1]
        archive_checkpoints = checkpoints

    transactions = get_transactions(o.transactions)
    check_transaction_sequences(transactions)

    filelist.append(get_last_transaction_log(o.transactions))

    filelist.extend(
        find_checkpoint_transactions(oldest_checkpoint, transactions)
    )

    filelist.append(get_latest_checkpoint(o.checkpoint))
    filelist.extend(archive_checkpoints)

    log("Oldest checkpoint to archive is: %s" % oldest_checkpoint)

    tar_cmd.extend(filelist)

    log("Files to archive:")
    for file in filelist:
        log("   %s" % file)

    code = 0
    if not o.dry_run:
        log("Archiving ...")
        code = subprocess.call(tar_cmd)

    sys.exit(code)