gpmovemirrors

#!/usr/bin/env python
# Line too long            - pylint: disable=C0301
# Invalid name             - pylint: disable=C0103
#
# Copyright (c) EMC Inc 2010. All Rights Reserved.
#
from gppylib.mainUtils import SimpleMainLock, ExceptionNoStackTraceNeeded

import copy
import datetime
import os
import sys
import socket
import signal
from time import strftime, sleep

try:
    from gppylib.commands.unix import *
    from gppylib.commands.gp import *
    from gppylib.commands.pg import PgControlData
    from gppylib.gparray import *
    from gppylib.gpparseopts import OptParser, OptChecker
    from gppylib.gplog import *
    from gppylib.db import dbconn
    from gppylib.userinput import *
    from gppylib.operations.startSegments import *
    from pygresql.pgdb import DatabaseError
    from pygresql import pg
    from gppylib import gparray, gplog, pgconf, userinput, utils
    from gppylib.parseutils import line_reader, parse_filespace_order, parse_gpmovemirrors_line, \
        canonicalize_address

except ImportError, e:
    sys.exit('ERROR: Cannot import modules.  Please check that you have sourced greenplum_path.sh.  Detail: ' + str(e))

# constants
MAX_BATCH_SIZE = 128

GPDB_STOPPED = 1
GPDB_STARTED = 2
GPDB_UTILITY = 3

GP_MOVEMIRROS_PID_FILE = "gpmovemirrors.pid"

description = ("""
Moves mirror segments in a pre-existing GPDB Array.
""")

EXECNAME = os.path.split(__file__)[-1]

_help = ["""
The input file should be be a plain text file with the format:

  filespaceOrder=[<filespace1_fsname>[:<filespace2_fsname>:...]
  <old_address>:<port>:<system_filespace_location> <new_address:port>:<replication_port>:<system_filespace_location>[:<fselocation>:...]

""",
         """
         An input file must be specified.
         """,
         ]


# ----------------------- Command line option parser ----------------------
def parseargs():
    parser = OptParser(option_class=OptChecker,
                       description=' '.join(description.split()),
                       version='%prog version $Revision$')
    parser.setHelp(_help)
    parser.remove_option('-h')

    parser.add_option('-i', '--input', dest="input_filename",
                      help="input expansion configuration file.", metavar="FILE")
    parser.add_option('-d', '--master-data-directory', dest='master_data_directory',
                      help='The master data directory for the system. If this option is not specified, \
                            the value is obtained from the $MASTER_DATA_DIRECTORY environment variable.')
    parser.add_option('-B', '--batch-size', dest='batch_size', type='int', default=16, metavar="<batch_size>",
                      help='Expansion configuration batch size. Valid values are 1-%d' % MAX_BATCH_SIZE)
    parser.add_option('-v', '--verbose', dest="verbose", action='store_true',
                      help='debug output.')
    parser.add_option('-l', '--log-file-directory', dest="logfile_directory",
                      help='The directory where the gpmovemirrors log file should be put. \
                            The default location is ~/gpAdminLogs.')
    parser.add_option('-C', '--continue', dest='continue_move', action='store_true',
                      help='Continue moving mirrors')
    parser.add_option('-h', '-?', '--help', action='help',
                      help='show this help message and exit.')
    parser.add_option('--usage', action="briefhelp")

    parser.set_defaults(verbose=False, filters=[], slice=(None, None))

    # Parse the command line arguments
    (options, args) = parser.parse_args()

    if len(args) > 0:
        logger.error('Unknown argument %s' % args[0])
        parser.exit()

    if options.batch_size < 1 or options.batch_size > 128:
        logger.error('Invalid argument.  -B value must be >= 1 and <= %s' % MAX_BATCH_SIZE)
        parser.print_help()
        parser.exit()

    if options.input_filename == None:
        logger.error('Missing argument. -i or --input is a required argument.')
        parser.print_help()
        parser.exit()

    try:
        if options.master_data_directory == None:
            options.master_data_directory = get_masterdatadir()
        options.gphome = get_gphome()
    except GpError, msg:
        logger.error(msg)
        parser.exit()

    if not os.path.exists(options.master_data_directory):
        logger.error('Master data directory does not exist.')
        parser.exit()

    options.pgport = int(os.getenv('PGPORT', 5432))

    return options, args


# -------------------------------------------------------------------------
def logOptionValues(options):
    """ """
    logger.info("Option values for this invocation of gpmovemirrors are:")
    logger.info("")
    logger.info("  --input                 = " + str(options.input_filename))
    logger.info("  --master-data-directory = " + str(options.master_data_directory))
    logger.info("  --batch-size            = " + str(options.batch_size))
    if options.verbose != None:
        logger.info("  --verbose               = " + str(options.verbose))
    if options.continue_move != None:
        logger.info("  --continue              = " + str(options.continue_move))
    logger.info("")


# -------------------------------------------------------------------------
pool = None


def shutdown():
    """used if the script is closed abrubtly"""
    logger.info('Shutting down gpmovemirrors...')
    if pool != None:
        pool.haltWork()
        pool.joinWorkers()


# -------------------------------------------------------------------------
def gpmovemirrors_status_file_exists(master_data_directory):
    """Checks if status exists"""
    return os.path.exists(master_data_directory + '/gpmovemirrors.status')


# -------------------------------------------------------------------------
def sig_handler(sig, arg):
    shutdown()

    signal.signal(signal.SIGTERM, signal.SIG_DFL)
    signal.signal(signal.SIGHUP, signal.SIG_DFL)

    # raise sig
    os.kill(os.getpid(), sig)


# -------------------------------------------------------------------------
def lookupGpdb(address, port, dataDirectory):
    """ Look up the segment gpdb by address, port, and dataDirectory """

    gpdbList = gparray.getDbList()
    for gpdb in gpdbList:
        if address == str(gpdb.getSegmentAddress()) and port == str(gpdb.getSegmentPort()) and dataDirectory == str(
                gpdb.getSegmentDataDirectory()):
            return gpdb
    return None


# -------------------------------------------------------------------------
def lookupReplicatonPort(gpdb):
    """ Look up the segment's replication port associated with the input parameters. """

    retValue = None
    if gpdb != None:
        retValue = gpdb.getSegmentReplicationPort()

    return retValue


# -------------------------------------------------------------------------
def lookupFilespaces(gpdb):
    """ Look up the segment's filespaces, and return map of filespace name : location. """

    retValue = {}
    gpFilespaceObjList = gparray.getFilespaces(includeSystemFilespace=False)
    gpdbFilespaceDict = gpdb.getSegmentFilespaces()

    for oid in gpdbFilespaceDict:
        if oid == SYSTEM_FILESPACE:
            continue
        name = gparray.getFileSpaceName(oid)
        location = gpdbFilespaceDict[oid]
        retValue[name] = location
    return retValue


# -------------------------------------------------------------------------
# -------------------------------------------------------------------------
class Mirror:
    """ This class represents information about a mirror. """

    def __init__(self, address, port, replicationPort, dataDirectory, filespaces):
        self.address = address
        self.port = port
        self.replicationPort = replicationPort
        self.dataDirectory = dataDirectory
        self.filespaces = filespaces

    def __str__(self):
        tempStr = "address = " + str(self.address) + '\n'
        tempStr = tempStr + "port = " + str(self.port) + '\n'
        tempStr = tempStr + "replicationPort = " + str(self.replicationPort) + '\n'
        tempStr = tempStr + "dataDirectory   = " + str(self.dataDirectory) + '\n'
        tempStr = tempStr + "filespaces      = " + str(self.filespaces)
        return tempStr


# -------------------------------------------------------------------------
# -------------------------------------------------------------------------
class Configuration:
    """ This class represents the mirrors. """

    def __init__(self):
        self.inputFile = None
        self.fileData = None
        self.filespaceNames = []
        self.oldMirrorList = []
        self.newMirrorList = []

    def read_input_file(self, inputFile):
        self.inputFile = inputFile

        fslist = None
        with open(inputFile) as f:

            for lineno, line in line_reader(f):
                if fslist is None:
                    fslist = parse_filespace_order(inputFile, lineno, line)
                    self.filespaceNames = fslist
                    continue

                rowMap, newFilespaces = parse_gpmovemirrors_line(inputFile, lineno, line, fslist)

                oldAddress = rowMap['oldAddress']
                oldPort = rowMap['oldPort']
                oldDataDirectory = rowMap['oldDataDirectory']
                oldGpdb = lookupGpdb(oldAddress, oldPort, oldDataDirectory)
                if oldGpdb == None or oldGpdb.getSegmentRole() != ROLE_MIRROR:
                    if oldGpdb == None:
                        raise Exception(
                            "Old mirror segment does not exist with given information: address = %s, port = %s, segment data directory = %s"
                            % (oldAddress, str(oldPort), oldDataDirectory))
                    else:
                        raise Exception(
                            "Old mirror segment is not currently in a mirror role: address = %s, port = %s, segment data directory = %s"
                            % (oldAddress, str(oldPort), oldDataDirectory))
                oldReplicationPort = lookupReplicatonPort(oldGpdb)
                oldFilespaces = lookupFilespaces(oldGpdb)
                oldMirror = Mirror(address=oldAddress
                                   , port=oldPort
                                   , replicationPort=oldReplicationPort
                                   , dataDirectory=oldDataDirectory
                                   , filespaces=oldFilespaces
                                   )
                self.oldMirrorList.append(oldMirror)
                newAddress = rowMap['newAddress']
                newPort = rowMap['newPort']
                newDataDirectory = rowMap['newDataDirectory']
                newReplicationPort = rowMap['newReplicationPort']
                newMirror = Mirror(address=newAddress
                                   , port=newPort
                                   , replicationPort=newReplicationPort
                                   , dataDirectory=newDataDirectory
                                   , filespaces=newFilespaces
                                   )
                self.newMirrorList.append(newMirror)

    def write_output_file(self, filename, oldToNew=True):
        """ Write out the configuration in a format appropriate for use with the -i option. """
        fp = open(filename, 'w')

        """ Set the first line up for the filespaceOrder entry. """
        fsString = ":".join(self.filespaceNames)
        line = "filespaceOrder=" + fsString
        line = line + '\n'
        fp.write(line)

        if oldToNew == True:
            firstList = self.oldMirrorList
            secondList = self.newMirrorList
        else:
            firstList = self.newMirrorList
            secondList = self.oldMirrorList
        for i in range(0, len(firstList)):
            oldEntry = firstList[i]
            newEntry = secondList[i]
            line = canonicalize_address(oldEntry.address) + \
                   ":" + str(oldEntry.port) + ":" + oldEntry.dataDirectory
            line = line + " "
            line = line + canonicalize_address(newEntry.address) + \
                   ":" + str(newEntry.port) + ":" + str(newEntry.replicationPort) + ":" + newEntry.dataDirectory
            for fsName in self.filespaceNames:
                line = line + ":" + newEntry.filespaces[fsName]
            line = line + '\n'
            fp.write(line)


# -------------------------------------------------------------------------
# --------------------------------- main ----------------------------------
# -------------------------------------------------------------------------
gp_movemirrors = None
remove_pid = True
options = None
args = None
pidfilepid = None  # pid of the process which has the lock
locktorelease = None
sml = None  # sml (simple main lock)

try:

    # setup signal handlers so we can clean up correctly
    signal.signal(signal.SIGTERM, sig_handler)
    signal.signal(signal.SIGHUP, sig_handler)

    logger = get_default_logger()

    options, args = parseargs()

    setup_tool_logging(appName=EXECNAME
                       , hostname=getLocalHostname()
                       , userName=getUserName()
                       , logdir=options.logfile_directory
                       )

    enable_verbose_logging()

    mainOptions = {"pidfilename": GP_MOVEMIRROS_PID_FILE}
    sml = SimpleMainLock(mainOptions)
    otherpid = sml.acquire()
    if otherpid is not None:
        logger.error("An instance of %s is already running (pid %s)" % ("gpmovemirrors", otherpid))
        sys.exit(1)

    logger.info("Invocation of gpmovemirrors mirrors")
    logOptionValues(options)

    dburl = dbconn.DbURL()

    # Get array configuration
    try:
        gparray = GpArray.initFromCatalog(dburl, utility=True)
    except DatabaseError, ex:
        logger.error('Failed to connect to database.  Make sure the')
        logger.error('Greenplum instance is running, and that')
        logger.error('your environment is correct, then rerun')
        logger.error('gpmovemirrors ' + ' '.join(sys.argv[1:]))
        sys.exit(1)

    """ Get the old to new mirror configuration. """
    newConfig = Configuration()
    newConfig.read_input_file(options.input_filename)

    """ Do some sanity checks on the input. """
    for mirror in newConfig.oldMirrorList:
        seg = lookupGpdb(mirror.address, mirror.port, mirror.dataDirectory)
        if seg == None:
            raise Exception(
                "Old mirror segment does not exist with given information: address = %s, port = %s, segment data directory = %s"
                % (mirror.address, str(mirror.port), mirror.dataDirectory))
        if seg.getSegmentContentId() == MASTER_CONTENT_ID:
            raise Exception(
                "Cannot move master or master mirror segments: address = %s, port = %s, segment data directory = %s"
                % (mirror.address, str(mirror.port), mirror.dataDirectory))
        if seg.getSegmentRole() != ROLE_MIRROR:
            raise Exception(
                "Old mirror segment is not currently in a mirror role: address = %s, port = %s, segment data directory = %s"
                % (mirror.address, str(mirror.port), mirror.dataDirectory))

    """ Create a backout file for the user if they choose to go back to the original configuration. """
    backout_filename = options.input_filename + "_backout_" + str(time.strftime("%m%d%y%H%M%S"))
    newConfig.write_output_file(backout_filename, False)

    """ Start gprecoverseg. """
    recoversegOptions = "-i " + newConfig.inputFile + " -v -a -d " + options.master_data_directory
    if options.logfile_directory != None:
        recoversegOptions = recoversegOptions + " -l " + str(options.logfile_directory)
    logger.info('About to run gprecoverseg with options: ' + recoversegOptions)
    cmd = GpRecoverSeg("Running gprecoverseg", options=recoversegOptions)
    cmd.run(validateAfter=True)

    """ Delete old mirror directories. """
    try:
        fileSpacesObjList = gparray.getFilespaces(includeSystemFilespace=False)
        totalDirsToDelete = len(newConfig.oldMirrorList) + len(fileSpacesObjList)
        numberOfWorkers = min(totalDirsToDelete, options.batch_size)
        pool = WorkerPool(numWorkers=numberOfWorkers)
        for mirror in newConfig.oldMirrorList:
            logger.info("About to remove old mirror segment directory: " + mirror.dataDirectory)
            cmd = RemoveFiles("remove old mirror segment directories"
                              , mirror.dataDirectory
                              , ctxt=REMOTE
                              , remoteHost=mirror.address
                              )
            pool.addCommand(cmd)
            for filespaceName in mirror.filespaces.keys():
                logger.info(
                    "About to remove old mirror filespace directory: " + filespaceName + " : " + mirror.filespaces[
                        filespaceName])
                cmd = RemoveFiles("remove old mirror segment directories"
                                  , mirror.filespaces[filespaceName]
                                  , ctxt=REMOTE
                                  , remoteHost=mirror.address
                                  )
                pool.addCommand(cmd)

        # Wait for the segments to finish
        try:
            pool.join()
        except:
            pool.haltWork()
            pool.joinWorkers()

        failure = False
        results = []
        for cmd in pool.getCompletedItems():
            r = cmd.get_results()
            if not cmd.was_successful():
                logging.error("Unable to remove old mirror segment directory: " + str(r))
                failure = True

        pool.haltWork()
        pool.joinWorkers()
        if failure:
            logging.error("Although the system is in the process of recovering the mirrors to their new location,")
            logging.error("There was an issue removing the old mirror segment directories.")
            raise Exception("Unable to complete removal of old mirror segment directories.")

    except Exception, e:
        raise e

except Exception, e:
    if options is not None and options.verbose:
        logger.exception("gpmovemirrors failed. exiting...")
    else:
        logger.error("gpmovemirrors failed: %s \n\nExiting..." % e)
    sys.exit(3)

except KeyboardInterrupt:
    # Disable SIGINT while we shutdown.
    signal.signal(signal.SIGINT, signal.SIG_IGN)

    shutdown()

    # Re-enabled SIGINT
    signal.signal(signal.SIGINT, signal.default_int_handler)

    sys.exit('\nUser Interrupted')

finally:
    if sml is not None:
        sml.release()