index : glusterfs.git	bug-deprecated dht-stale-layout-fixes exp experimental heal-info master release-2.0 release-3.0 release-3.1 release-3.10 release-3.11 release-3.12 release-3.13 release-3.2 release-3.3 release-3.4 release-3.5 release-3.6 release-3.7 release-3.8 release-3.8-fb release-3.9 release-4.0 release-4.1 release-5 release-6 release-7 release-8 round-robin2 testing-regression-job

summaryrefslogtreecommitdiffstats

log msg author committer range

path: root/AUTHORS

	Commit message (Expand)	Author	Age	Files	Lines
*	cleaned up some outdated files	Amar Tumballi	2010-03-12	1	-3/+0
*	Added all files	Vikas Gorur	2009-02-18	1	-0/+3

generated by cgit (git 2.34.1) at 2026-01-06 17:50:58 +0000

#!/usr/bin/env python

# Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com/>
# This file is part of GlusterFS.
#
# This file is licensed to you under your choice of the GNU Lesser
# General Public License, version 3 or any later version (LGPLv3 or
# later), or the GNU General Public License, version 2 (GPLv2), in all
# cases as published by the Free Software Foundation.

import sys
from errno import ENOENT
import time
from multiprocessing import Process
import os
import xml.etree.cElementTree as etree
from argparse import ArgumentParser, RawDescriptionHelpFormatter, Action
import logging
import shutil

from utils import execute, is_host_local, mkdirp, fail
from utils import setup_logger, human_time
import conf


PROG_DESCRIPTION = """
GlusterFS Incremental API
"""
ParseError = etree.ParseError if hasattr(etree, 'ParseError') else SyntaxError

logger = logging.getLogger()


class StoreAbsPath(Action):
    def __init__(self, option_strings, dest, nargs=None, **kwargs):
        super(StoreAbsPath, self).__init__(option_strings, dest, **kwargs)

    def __call__(self, parser, namespace, values, option_string=None):
        setattr(namespace, self.dest, os.path.abspath(values))


def get_pem_key_path(session, volume):
    return os.path.join(conf.get_opt("session_dir"),
                        session,
                        volume,
                        "%s_%s_secret.pem" % (session, volume))


def node_run(volume, host, path, start, outfile, args, fallback=False):
    """
    If host is local node, execute the command locally. If not local
    execute the CHANGE_DETECTOR command via ssh and copy the output file from
    remote node using scp.
    """
    localdir = is_host_local(host)
    pem_key_path = get_pem_key_path(args.session, args.volume)

    # If Full backup is requested or start time is zero, use brickfind
    change_detector = conf.get_change_detector(args.change_detector)
    if ((start == 0 or args.full) and args.change_detector == "changelog") or \
       fallback:
        change_detector = conf.get_change_detector("brickfind")

    # CHANGE_DETECTOR <SESSION> <VOLUME> <BRICK> <OUTFILE> <START> --debug
    # --gfidpath <TYPE>
    cmd = [change_detector,
           args.session,
           volume,
           path,
           outfile,
           str(start),
           "--output-prefix",
           args.output_prefix] + \
        (["--debug"] if args.debug else []) + \
        (["--full"] if args.full else [])

    if not localdir:
        # prefix with ssh command if not local node
        cmd = ["ssh",
               "-i", pem_key_path,
               "root@%s" % host] + cmd

    rc, out, err = execute(cmd, logger=logger)
    if rc == 2:
        # Partial History Fallback
        logger.info("%s %s Fallback to brickfind" % (host, err.strip()))
        # Exit only from process, handled in main.
        sys.exit(rc)
    elif rc != 0:
        fail("%s - Change detection failed" % host, logger=logger)

    if not localdir:
        cmd_copy = ["scp",
                    "-i", pem_key_path,
                    "root@%s:/%s" % (host, outfile),
                    os.path.dirname(outfile)]
        execute(cmd_copy, exit_msg="%s - Copy command failed" % host,
                logger=logger)


def node_cleanup(host, args):
    localdir = is_host_local(host)

    pem_key_path = get_pem_key_path(args.session, args.volume)

    # CHANGE_DETECTOR <SESSION> <VOLUME> <BRICK> <OUTFILE> <START> --debug
    # --gfidpath <TYPE>
    cmd = [conf.get_opt("nodecleanup"),
           args.session,
           args.volume] + (["--debug"] if args.debug else [])

    if not localdir:
        # prefix with ssh command if not local node
        cmd = ["ssh",
               "-i", pem_key_path,
               "root@%s" % host] + cmd

    execute(cmd, exit_msg="%s - Cleanup failed" % host, logger=logger)


def cleanup(nodes, args):
    pool = []
    for num, node in enumerate(nodes):
        host, brick = node[1].split(":")
        # temp output file
        node_outfile = os.path.join(conf.get_opt("working_dir"),
                                    args.session,
                                    args.volume,
                                    "tmp_output_%s.txt" % num)

        try:
            os.remove(node_outfile)
        except (OSError, IOError):
            # TODO: Cleanup Failure, Handle
            pass

        p = Process(target=node_cleanup,
                    args=(host, args))
        p.start()
        pool.append(p)

    exit_codes = 0
    for p in pool:
        p.join()
        exit_codes += (0 if p.exitcode == 0 else 1)

    if exit_codes != 0:
        sys.exit(1)


def failback_node_run(brick_details, idx, volume, start, outfile, args):
    host, brick = brick_details.split(":")
    p = Process(target=node_run,
                args=(volume, host, brick, start, outfile, args, True))
    p.start()
    p.join()
    return p.exitcode


def run_in_nodes(volume, start, args):
    """
    Get nodes of volume using gluster volume info, spawn a process
    each for a Node. Merge the output files once all the process
    complete their tasks.
    """
    nodes = get_nodes(volume)
    pool = []
    node_outfiles = []
    for num, node in enumerate(nodes):
        host, brick = node[1].split(":")
        # temp output file
        node_outfile = os.path.join(conf.get_opt("working_dir"),
                                    args.session,
                                    volume,
                                    "tmp_output_%s.txt" % num)
        node_outfiles.append(node_outfile)
        p = Process(target=node_run, args=(volume, host, brick, start,
                                           node_outfile, args))
        p.start()
        pool.append(p)

    exit_codes = 0
    for idx, p in enumerate(pool):
        p.join()
        # Handle the Changelog failure, fallback to Brickfind
        if p.exitcode == 2:
            rc = failback_node_run(nodes[idx][1], idx, volume, start,
                                   node_outfiles[idx], args)
            exit_codes += (0 if rc == 0 else 1)
        elif p.exitcode != 0:
            exit_codes += (0 if p.exitcode == 0 else 1)

    if exit_codes != 0:
        sys.exit(1)

    # Merge all output files
    cmd = ["sort", "-u"] + node_outfiles + ["-o", args.outfile]
    execute(cmd,
            exit_msg="Failed to merge output file