#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
gpmigrator_mirror [options] old-gphome new-gphome

Options:
  -h, --help            show this help message and exit
  -v, --version         show the program's version number and exit
  -q                    quiet mode
  -d DIRECTORY          the master host data directory
  -l DIRECTORY          log file directory
  --debug               debugging information
'''

#============================================================
import sys, os

# Python version 2.6.2 is expected, must be between 2.5-3.0
if sys.version_info < (2, 5, 0) or sys.version_info >= (3, 0, 0):
    sys.stderr.write("Error: %s is supported on Python versions 2.5 or greater\n" 
                     "Please upgrade python installed on this machine." 
                     % os.path.split(__file__)[-1])
    sys.exit(1)

try:
    from gppylib.operations.gpMigratorUtil import *
except ImportError, e:
    sys.exit('Error: unable to import module: ' + str(e))

libdir = os.path.join(sys.path[0], 'lib/')

logger        = get_default_logger()
EXECNAME      = os.path.split(__file__)[-1]
MIGRATIONUSER = 'gpmigrator'
LOCKEXT       = '.gpmigrator_orig'
WORKDIR       = 'gpmigrator'
BACKUPDIR     = 'backup'
UPGRADEDIR    = 'upgrade'
PARALLELISM   = 16




#============================================================
__version__ = '$Revision: #1 $'




#============================================================
def makeCommand(oldHome, newHome,
                 command, dataDirectories, option1):
    # space separated list of directories
    datadirs = base64.urlsafe_b64encode(pickle.dumps(dataDirectories))
    cmd = [
           EXECNAME,
           oldHome, 
           newHome,
           '--internal-command=' + str(command),
           '--internal-dirs=' + urllib.quote(datadirs),
           '--internal-option=' + urllib.quote(str(option1)),
           '--debug',
           '--quiet',
           ]
    return cmd

#============================================================
def logGenericErrorMessage():
    logger.fatal('Upgrade Failed..!!!')
    logger.fatal('...Database might be in an unusable state')
    logger.fatal('...Please check the logs, resolve the issue an re-run gpmigrator_mirror')


#============================================================
class GPUpgrade(GPUpgradeBase):
    '''
    Greenplum Database Upgrade Utility
    '''

    # INTERNAL command:
    #   MASTER    - default, run on master, [not included in list]
    #   CHKDOWN   - check the db is down
    #   SETCATVERSION - set catalog version
    commands = ['CHKDOWN', 'SETCATVERSION']

    #------------------------------------------------------------
    def __init__(self):
        '''
        The most basic of initialization
        '''
        super(GPUpgrade, self).__init__()
        
        # Environment and system info
        self.datadirs  = None     # Set of all data and mirror directories (Segment only)

        
        # Mirrored Upgrade State
        self.mirror_upgrade_state = None
        self.mirror_upgrade_state_file = None

    #------------------------------------------------------------
    def Setup(self):
        '''
        Basic initialization, separate from __init__ for exception
        handling purposes.
        '''

        # GPHOME, PYTHONPATH must be setup properly
        #    GPHOME must match the location of this file
        #    The first PYTHONPATH must be based on the proper gphome 
        gphome_bin = os.path.realpath(os.path.split(__file__)[0])
        gphome     = os.path.split(gphome_bin)[0]
        env_GPHOME = os.path.realpath(os.environ.get('GPHOME'))
        if (env_GPHOME != gphome):
            logger.fatal(" $GPHOME is set to %s which is not newhome" % env_GPHOME)
            logger.fatal(' source the greenplum.sh from the newhome to setup env ')
            raise UpgradeError('Initialization failed')

        pythonpath     = os.path.join(gphome, "lib", "python")
        env_PYTHONPATH = os.path.realpath(os.environ.get('PYTHONPATH').split(':')[0])
        if (env_PYTHONPATH != pythonpath):
            logger.fatal(' $PYTHONPATH is incorrect ')
            logger.fatal(' source the greenplum.sh from the newhome to setup env ')
            raise UpgradeError('Initialization failed')

        # This is the same path used by gpinitsystem
        self.path = '/usr/kerberos/bin:/usr/sfw/bin:/opt/sfw/bin'
        self.path += ':/usr/local/bin:/bin:/usr/bin:/sbin:/usr/sbin:/usr/ucb'
        self.path += ':/sw/bin'

        # Set defaults
        self.user      = os.environ.get('USER') or os.environ.get('LOGNAME')
        self.host      = self.RunCmd('hostname')
        self.masterdir = os.environ.get('MASTER_DATA_DIRECTORY')
        
        self.ParseInput()
        
        # Setup worker pool
        self.pool = base.WorkerPool(numWorkers=PARALLELISM);

        # Extract path from input master directory
        if self.cmd == 'MASTER':
            logger.info('Beginning upgrade')
            logger.info('Checking configuration')

            if not self.masterdir or len(self.masterdir) == 0:
                raise UpgradeError('MASTER_DATA_DIRECTORY is not defined')
            self.masterdir = self.masterdir.rstrip('/')

            # The configuration file
            conf = os.path.join(self.masterdir, 'postgresql.conf')

            # Simple function to look for settings in the conf file
            def getconf(x):
                conf_re  = re.compile('^\s*%s\s*=\s*(\w+)' % x)
                try:
                    conf_str = self.RunCmd('grep %s %s' % (x, conf))
                except CmdError:
                    conf_str = ""     # grep returns errorcode on no match
                value = None
                for line in conf_str.split('\n'):
                    match = conf_re.search(line)
                    if match:
                        value = match.group(1)
                return value

            # Find the port for this segment:
            self.masterport = getconf('port')
            if self.masterport == None:
                raise UpgradeError('Could not determine master port from ' + conf)
            self.masterport = int(self.masterport)

            self.sock_dir = getconf('unix_socket_directory')
            if not self.sock_dir:
                self.sock_dir = '/tmp/'

            # Verify that (max_connections == superuser_reserved_connections)
            # max_conn = getconf('max_connections')
            # reserved = getconf('superuser_reserved_connections')

            masterpath, masterdir = os.path.split(self.masterdir)
            self.workdir   = os.path.join(self.masterdir, WORKDIR)
            self.oldenv    = self.SetupEnv(self.oldhome, self.masterdir)
            self.newenv    = self.SetupEnv(self.newhome, self.masterdir)

        else:
            self.oldenv = self.SetupEnv(self.oldhome, None)
            self.newenv = self.SetupEnv(self.newhome, None)

    #------------------------------------------------------------
    def ParseInput(self):
        '''
        Parses and validates input to the script
        '''

        try:
            parser = optparse.OptionParser(usage=(cli_help(EXECNAME) or __doc__), add_help_option=False)
            parser.add_option('-v', '--version', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-h', '--help', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-d', dest='directory', 
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-l', dest='logdir',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-c', '--check-only', dest='checkonly', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-q', '--quiet', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--debug', action='store_true',
                              help=optparse.SUPPRESS_HELP)

            parser.add_option('--internal-command', dest='command',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--internal-dirs', dest='datadirs',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--internal-option', dest='option',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--skip-checkcat', dest='skipcheckcat',
                              action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--fault-injection', dest='faultinjection',
                              type='int', help=optparse.SUPPRESS_HELP)
            (options, args) = parser.parse_args()
            
            if options.version:
                print EXECNAME + ' ' + __version__
                sys.exit(0)

            if options.help:
                usage(EXECNAME)
                sys.exit(0)

            if len(args) != 2:
                usage(EXECNAME)
                msg = "incorrect number of arguments"
                if len(args) > 0:
                    msg += ": %s" % str(args)
                parser.error(msg)
        except Exception, e:
            usage(EXECNAME)
            raise UpgradeError('Error parsing input: ' + str(e))

        if options.directory:
            self.masterdir = options.directory
        if options.command:
            self.cmd = options.command
        self.oldhome   = args[0].rstrip('/')
        self.newhome   = args[1].rstrip('/')
        if options.datadirs:
            self.datadirs = pickle.loads(base64.urlsafe_b64decode((urllib.unquote(options.datadirs))))
        if options.option:
            self.option   = urllib.unquote(options.option)
        if options.checkonly:
            self.checkonly = True
        if options.debug:
            self.debug = True
            enable_verbose_logging()
        if options.quiet:
            quiet_stdout_logging()
        if options.logdir:
            self.logdir = options.logdir
        else:
            self.logdir = os.path.join(os.environ['HOME'], 'gpAdminLogs')

        try:
            setup_tool_logging(EXECNAME,
                               unix.getLocalHostname(),
                               unix.getUserName(),
                               self.logdir)
        except OSError, e:
            logger.fatal('cannot log to %s: %s' % (self.logdir, str(e)))
            exit(1)
        
        self.skipcheckcat = options.skipcheckcat
        self.faultinjection = options.faultinjection

        if self.cmd == 'MASTER':

            today = date.today().strftime('%Y%m%d')
            if not os.path.isdir(self.logdir):
                os.makedirs(self.logdir, 0700)
            logname = os.path.join(self.logdir, '%s_%s.log' % (EXECNAME, today))
            self.logfile = open(logname, 'a')

            if self.masterdir == None:
                logger.fatal('MASTER_DATA_DIRECTORY parameter not set')
                raise UpgradeError('Initialization failed')
            if not os.path.exists(self.masterdir):
                logger.fatal('MASTER_DATA_DIRECTORY: %s not found'
                             % self.masterdir)
                raise UpgradeError('Initialization failed')
            if not os.path.isabs(self.masterdir):
                self.masterdir = os.path.abspath(self.masterdir)
            if not os.path.exists(self.oldhome):
                logger.fatal(' directory not found: ' + self.oldhome)
                raise UpgradeError('Initialization failed')
            if not os.path.exists(self.newhome):
                logger.fatal(' directory not found: ' + self.newhome)
                raise UpgradeError('Initialization failed')
            if not os.path.isabs(self.oldhome):
                self.oldhome = os.path.abspath(self.oldhome)
            if not os.path.isabs(self.newhome):
                self.newhome = os.path.abspath(self.newhome)


    #------------------------------------------------------------
    def CallSlaves(self, cmd, option=''):
        '''
        Calls every host to execute the given command with the given option 
        value
        '''
  
        logger.debug("Remote call: %s" % cmd)

        # Check for things that should never happen
        if self.cmd != 'MASTER':
            raise UpgradeError("Recursive communication error")
        if not self.array:
            raise UpgradeError("Failure initializing array")
        if not self.hostcache:
            raise UpgradeError("Failure initializing host cache")
        if not self.pool:
            raise UpgradeError("Failure initializing worker pool")
        
        # Construct the commands to pass to the worker pool
        hosts = self.hostcache.get_hosts()
        for host in hosts:
            hostname = host.hostname

            # Skip any hosts in the cache that contain no segments for this
            # configuration.
            if len(host.dbs) == 0:
                continue

            # Get the data directories for this host:
            datadirs = []
            for seg in host.dbs:
                datadirs.append(seg.getSegmentDataDirectory())
            
            # Skip any hosts that have no applicable data directories
            if len(datadirs) == 0:
               continue

            cmdList = makeCommand(oldHome=self.oldhome,
                             newHome=self.newhome,
                             command=cmd,
                             dataDirectories=datadirs,
                             option1=option)

            c = GpUpgradeCmd("gpmigrator_mirror remote call",
                             cmdList,
                             ctxt=base.REMOTE,
                             remoteHost=hostname)
            self.pool.addCommand(c)

        # Wait for the segments to finish
        try:
            self.pool.join()
        except:
            self.pool.haltWork()
            self.pool.joinWorkers()
            
        failure = False
        results = []
        for cmd in self.pool.getCompletedItems():
            r = cmd.get_results()

            # Going through the gppylib Command interface all stderr from the
            # remote calls gets redirected to stdout, which is unfortunate
            # because we'd like to be able to differentiate between the two.
            #
            # We keep the stdout chatter to a minimum by passing --quiet to
            # the remote calls which performs quiet stdout logging.

            # sys.stderr.write(r.stderr)
            msg = r.stdout.strip()
            results.append(msg)
            
            if not cmd.was_successful():
                log_literal(logger, logging.ERROR, msg)
                failure=True

        if failure:
            raise UpgradeError("Fatal Segment Error")

        # Warning this output contains everything written to stdout,
        # which unfortunately includes some of the logging information
        return "\n".join(results)


    #------------------------------------------------------------
    def CheckVersions(self):
        '''
        Validates that the upgrade from old->new is okay
        '''

        # Log the OS type info
        os_type = os.uname()[0]
        os_ver  = os.uname()[2]
        logger.info('Operating System: %s %s' % (os_type, os_ver))

        # Log version checking
        logger.info('Checking version compatibility')

        # If we got a version, but it isn't recognized by the GpVersion class
        # then it is likely not a Greenplum database.
        oldversion = self.getversion(self.oldhome, self.oldenv)
        newversion = self.getversion(self.newhome, self.newenv)
        try:
            oldversion = GpVersion(oldversion)
        except:
            raise UpgradeError('Source not a Greenplum Database: ' + oldversion)
        try:
            newversion = GpVersion(newversion)
        except:
            raise UpgradeError('Target not a Greenplum Database: ' + newversion)

        logger.info('Source Version: (Greenplum Database) %s' % str(oldversion))
        logger.info('Target Version: (Greenplum Database) %s' % str(newversion))

        if newversion == oldversion:
            raise UpgradeError("Greenplum Database is already version '%s'" 
                               % str(newversion))
        elif newversion.isVersionRelease(oldversion):
            raise UpgradeError("Upgrade not needed to go from version '%s' to version '%s'" 
                               % (str(oldversion), str(newversion)))

        is_supported_version(oldversion)
        is_supported_version(newversion)

        # We don't support downgrade
        if newversion < oldversion:
            raise UpgradeError("Downgrade to Greenplum Database %s not supported from gpmigrator" 
                               % str(newversion))

        if not newversion.isVersionCurrentRelease():
            main = GpVersion('main')
            raise UpgradeError(
                "Upgrade from '%s' to '%s' not supported.  Target version should be Greenplum Database %s"
                % (str(oldversion), str(newversion), main.getVersionRelease()))

        # Compare old version catalog number with the catalog in 
        # MASTER_DATA_DIRECTORY
        catalog_re = re.compile('Catalog version number: *(.*)')
        oldcat = self.RunCmd('postgres --catalog-version', env=self.oldenv)
        logger.info("Source %s" % oldcat)
        newcat = self.RunCmd('postgres --catalog-version', env=self.newenv)
        logger.info("Target %s" % newcat)

        control = self.RunCmd('pg_controldata ' + self.masterdir, env=self.oldenv)
        for line in control.split('\n'):
            m = catalog_re.match(line)
            if m:
                logger.info("Data   %s " % line)
                if (line != oldcat):
                    logger.debug('catalog mismatch: expected %s, found %s'
                             % (oldcat, line))
                    msg = 'Catalog in %s does not match source binary' % self.masterdir
                    raise UpgradeError(msg)
                break

        # For the moment everything goes:
        #   - Additional checks will be made once the old database is up
        logger.info('Versions are compatible')        
               

    #------------------------------------------------------------
    def GenerateConfigFiles(self):
        '''
        Creates 'gp_databases' 'gp_array_config' and 'gp_config' files
        '''

        logger.info('Generating config files')        

        # Write a dump of the databases hash
        db_file = os.path.join(self.workdir, 'gp_databases')
        o = open(db_file, 'w')
        for (oid, name) in self.dbs.iteritems():
            o.write('%s:%s\n' % (oid, name))
        o.close()

        # Write a dump of the gparray object
        array_file = os.path.join(self.workdir, 'gp_array_config')
        self.array.dumpToFile(array_file)

        logger.info('Configuration files generated')


    #------------------------------------------------------------
    def ReadInfo(self):
        ''' 
        When restarting an upgrade this method reads the configuration
        information from the cached status files.  

        It corresponds directly with ExtractInfo() which gathers the same
        information from an active database.

        It is also the inverse of GenerateConfigFiles() which creates the files
        that this function reads.

        In normal processing usually ExtractInfo() is called rather than
        ReadInfo().  This function is used /instead/ when we are doing Revert()
        processing, or when resuming an upgrade.  E.g. cases when we do not
        want to start the database to get the information we need because the
        database may not be in a stable state.
        '''

        # self.array - configuration information
        array_config = os.path.join(self.workdir, 'gp_array_config')
        if not os.path.exists(array_config):
            raise UpgradeError("  '%s' not found" % array_config)
        self.array = GpArray.initFromFile(array_config)

        # self.dbs - database information
        db_file = os.path.join(self.workdir, 'gp_databases')
        if not os.path.exists(db_file):
            raise UpgradeError("  '%s' not found" % db_file)
        f = open(db_file, 'r')
        for line in f:
            (oid, name) = line.split(':', 1)
            self.dbs[oid] = name.strip()
        f.close()

        # self.hostcache - cache of hostnames
        self.hostcache = GpHostCache(self.array, self.pool)
        failed_pings = self.hostcache.ping_hosts(self.pool)
        if len(failed_pings) > 0:
            raise UpgradeError(
                "Cannot upgrade while there are unreachable hosts")

        # self.mirrors
        dbs = self.array.getSegDbList()
        self.mirrors  = filter(GpDB.isSegmentMirror, dbs)
                
        # Setup the master catalog dirs
        self.mastercatdir = self.array.master.catdirs



    #------------------------------------------------------------
    def ExtractInfo(self):
        '''
        Get all the information we need from the old instance
        '''

        # Can only extract with a running database
        env = self.CheckUp()

        # -- Get the array information
        logger.info('Extracting configuration information')
        port  = self.masterport
        user  = env['USER']
        url   = dbconn.DbURL(port=port, dbname='template1', username=user)
        array = GpArray.initFromCatalog(url, utility=True)
        self.array = array

        # -- Determine if there are any invalid segments
        logger.info('Checking that all segments are valid')
        invalid = array.get_invalid_segdbs()
        if len(invalid) > 0:
            for db in invalid:
                logger.fatal('INVALID SEGMENT: %s:/%s' % 
                             (db.getSegmentHostName(),
                              db.getSegmentDataDirectory()))
            raise UpgradeError('Cannot upgrade database with invalid segments')


        # -- Get a list of available databases, note this includes "template0"!
        logger.info('Creating list of databases')
        databases = self.Select("SELECT oid, datname from pg_database")
        for row in databases:
            self.dbs[str(row['oid'])] = row['datname']

        # -- Older releases didn't have hostname as part of configuration
        # make sure we have an acurate list of address->host lookups.
        logger.info('Validating hosts')
        self.hostcache = GpHostCache(self.array, self.pool)
        for host in self.hostcache.get_hosts():
            for seg in host.dbs:
                seg.setSegmentHostName(host.hostname)
        failed_pings = self.hostcache.ping_hosts(self.pool)
        if len(failed_pings) > 0:
            raise UpgradeError("Cannot upgrade while there are unreachable "
                               "hosts")
            
        # Setup the master catalog dirs
        self.mastercatdir = self.array.master.catdirs
            
        master = array.master
        standbymaster = array.standbyMaster

        # Oddly gparray doesn't have methods for fetching primary segments
        # specifically
        dbs = array.getSegDbList()
        segments = filter(GpDB.isSegmentPrimary, dbs)
        self.mirrors  = filter(GpDB.isSegmentMirror, dbs)

        # Internal sanity checking
        if len(segments) == 0:
            raise UpgradeError('No segments found')
        if standbymaster > 0:
            raise UpgradeError('Cannot upgrade while standbymaster is running')

        # Check for unsupported index types.
        found = False
        logger.info("Validating indexes")
        oids = sorted(self.dbs.keys())
        for dboid in oids:
            db = self.dbs[dboid]

            if db == "template0": 
                continue
            indexes = self.Select('''
               SELECT quote_ident(n.nspname) || '.' || 
                      quote_ident(c.relname) as index, 
                      m.amname as kind
               FROM   pg_class c 
                 join pg_namespace n on (c.relnamespace = n.oid) 
                 join pg_am m on (c.relam = m.oid) 
               WHERE  c.relkind = 'i' and m.amname in ('gin', 'hash')
            ''', db=db)
            if len(indexes) > 0:
                if found == False:
                    logger.fatal("Unable to upgrade the following indexes")
                    found = True
                logger.fatal("  Database: %s" % db)
                for row in indexes:
                    logger.fatal("     %s  [%s]" % (row['index'], row['kind']))
        if found:
            raise UpgradeError("Deprecated index types must be removed prior "
                               "to upgrade")

        # Check for SQL_ASCII database encoding
        logger.info("Validating database encoding")
        encoding = self.Select("SELECT datname FROM pg_database WHERE encoding=0")
        if len(encoding) > 0:
            logger.error("Deprecated database encodings found:")
            for datname in encoding:
                logger.error("    %s  [SQL_ASCII]" % datname)
            raise UpgradeError("Deprecated database encodings found - contact "
                               "Greenplum Support")

        logger.info('Configuration acquired')
                    
    #------------------------------------------------------------
    def ExecuteSQLFileOnAllDBs(self, runfile):
        '''
        Execute "runfile" in upgrade mode on all databases, including template0
        '''
        def transform_db(dbname, dboid, tfile):
            '''
            Run the script on the specified database
            '''
            logger.debug("Executing on db = %s dboid = %s tfile = %s" % \
                     (str(dbname), str(dboid), tfile))

            # Execute the upgrade script
            cmd = ' '.join(["PGOPTIONS='-c gp_maintenance_conn=true'", 'psql', '-f', tfile, dbname, self.user])
            logger.debug('running ' + cmd)
            p = subprocess.Popen(cmd, shell = True,
                                  close_fds = True,
                                  stdout=subprocess.PIPE, 
                                  stderr=subprocess.PIPE, 
                                  env=self.newenv)
            result = p.communicate()
        
            if p.returncode == 0:
                logger.debug('successfully transformed ' + dbname)
                logger.debug(result[0].strip())
                logger.debug(result[1].strip())
            else:
                logger.debug(result[0])
                raise CmdError(str(cmd), result[0], result[1])


        logger.info('Performing catalog transformation')
        # MPP-10166 - because some of our transformations involve creates
        # of views with unstable oids we must make sure that we process 
        # the databases in exactly the same order on every segment, so
        # sorting the oids is essential.
        oids = sorted(self.dbs.keys())
        for dboid in oids:
            db = self.dbs[dboid]
                
            logger.info('... ' + db)
            transform_db(db, dboid, runfile)
            # Test failure during transformation
            if self.faultinjection == 1:
                raise UpgradeError("faultinjection=%d" % self.faultinjection)

    #------------------------------------------------------------
    def SetCatVersion(self, tonewver):
        '''
        Upgrade/downgrade pg_control file to another version. Cluster should
        be clean shutdown.
        '''

        logger.info('Modifying catalog version')
        
        if (tonewver==True):
            curenv     = self.newenv
            curhome    = self.newhome
        else:
            curenv     = self.oldenv
            curhome    = self.oldhome
        toversion = GpVersion(self.getversion(curhome, curenv))
        to        = toversion.getVersionRelease()

        logger.info("To %s"%to)
        self.CheckDown()

        def get_control_data(datadir):
             '''
             Parse the output of pg_controldata run on data directory, returning
             catalog version and state
             '''
             cmd = ' '.join(['pg_controldata', datadir])
             p = subprocess.Popen(cmd, shell=True, close_fds=True,
                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                  env=curenv)
             result = p.communicate()
        
             if p.returncode != 0:
                 raise CmdError(cmd, result[0], result[1])
        
             out = result[0].strip()
             ver = ""
             state = ""
             for line in out.split('\n'):
                  s = line.split(':')
                  if s[0] == 'Catalog version number':
                        ver = s[1].strip()
                  elif s[0] == 'Database cluster state':
                        state = s[1].strip()
        
             return [ver, state]
        
        def setcatversion(datadir, to):
            '''
            Set catalog version to 'to'

            We use gpmodcontrol to update pg_control file since in this
            version (4.3) we change the ControlFile structure.  If only
            catalog_version_no needs to change, we could use gpmodcatversion
            as was previously.  In case of gpmodcontrol (but not the case
            for gpmodcatversion), the newer environment is responsible to
            take care of downgrade too since older version does not have
            knowledge of the newer structure.
            '''
            # It is an error if the current state is not "shut down", that
            # means that there may be pending xlog records which will cause 
            # problems for the upgrade.
            (ver, state) = get_control_data(datadir)
            if state != "shut down":
                raise UpgradeError("Cannot upgrade: Database did not shutdown cleanly")

            cmd = ['%s/bin/lib/gpmodcontrol' % self.newhome]
            if not tonewver:
                cmd.append('--downgrade')
            cmd.append(datadir)
            p = subprocess.Popen(cmd,
                                 shell=False,
                                 close_fds=True,
                                 env=self.newenv, # always newenv
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)
            result = p.communicate()
            if p.returncode != 0:
                raise Exception("could not update catalog to %s" % to)


        if self.cmd == 'MASTER':
            datadirs = [ self.masterdir ]
            self.CallSlaves('SETCATVERSION', tonewver)
            if self.faultinjection == 2:
                raise UpgradeError("faultinjection=%d" % self.faultinjection)
        else:
            datadirs = self.datadirs

        for dir in datadirs:
            setcatversion(dir, to)
            if self.faultinjection == 3:
                raise UpgradeError("faultinjection=%d" % self.faultinjection)
            

        
    #------------------------------------------------------------
    def Run(self):
        """
        The main execution method - switches between submodes.

        The main frontend will execute the main PerformUpgrade() function,
        during which it will execute CallSlaves with various backend operations.
        For each one of the main backend operations we will spawn the backend
        executables, this switches between each of these operations.
        """

        # Execution on the master
        if self.cmd == 'MASTER':
            self.PerformMirroredUpgrade()
        
        # The rest of the options are executed on remote hosts
        elif self.cmd == 'CHKDOWN':
            self.CheckDown()
        elif self.cmd == 'SETCATVERSION':
            self.SetCatVersion((str(self.option)==str(True)))
        else:
            raise Exception('Unknown cmd: ' + str(self.cmd))
    
        if self.pool:
            t = self.pool
            self.pool = None
            del t

    # MIRROR UPGRADE CODE
    #------------------------------------------------------------
    def ForceShutDown(self):
        # The cleanest shutdown is to start the master in master only
        # and then do gpstop -a.
        # This simple thing couldn't shutdown segment if the master is dead.
        try:
            env = self.CheckUp()
            self.dbup = [env, False]
            self.Shutdown()
        except:
            pass


    def CreateMirrorWorkDir(self):
        # Create the workdir
        #   if the dir exists, clean it first
        self.RunCmd('rm -rf ' + self.workdir)
        os.mkdir(self.workdir)
        return "ExtractInfoAndCopyMaster"

    #------------------------------------------------------------
    def ExtractInfoAndCopyMaster(self):
        """
        Extract the database info and ceck for upgrade pre-requisite
        Serialize the db info to a flat file for "resume" (if crashed)
        Create a Master Copy
        """
        self.ExtractandWriteInfo()
        self.CreateMasterCopy()
        return "TurnOffMirror"
    
    def ExtractandWriteInfo(self):
        self.CheckVersions()
        self.ForceShutDown()
        self.Startup(self.oldenv)

        # Pre-requisite check
        #   1. all segments must be up
        #.  2. all mirror and segments must be in sync
        #   3. all mirror/segment must be in their preferred role
        #   4. it must have mirror
        #   5. pass gpcheckcat
        
        # This query check 1, 2 and 3.
        nonsynccnt = self.Select('''select count(*)
             from pg_catalog.gp_segment_configuration
             where status != 'u' or mode != 's' or role != preferred_role
            ''' )
        if (nonsynccnt[0] > 0):
            raise UpgradeError("Mirror or Segments are down or out of sync")

        # Now, extract the info before going on with more checks
        self.ExtractInfo()
        
        #   4. it must have mirror
        if (len(self.mirrors) ==0):
            raise UpgradeError("This upgrade requires mirroring.")
        
        #   5. pass common pre-upgrade check
        self.PreUpgradeCheck()
        
        # If we're in check only mode, then we exit here
        if self.checkonly:
            sys.exit(0)
        
        # Everything is ok. Write info to disk
        self.GenerateConfigFiles()

    #------------------------------------------------------------
    def do_copy(self, src, dst, f):
        '''
        Copy a directory tree
        '''
        if os.path.isdir(os.path.join(src, f)):
            if os.path.exists(os.path.join(dst,f)):
                logger.debug('rmtree(%s)' % os.path.join(dst,f))
                shutil.rmtree(os.path.join(dst,f))
            logger.debug('cptree(%s)' % os.path.join(dst,f))
            shutil.copytree(os.path.join(src, f), os.path.join(dst, f))
        else:
            if os.path.exists(os.path.join(dst,f)):
                logger.debug('rm(%s)' % os.path.join(dst,f))
                os.remove(os.path.join(dst,f))
            logger.debug('cp(%s)' % os.path.join(dst,f))
            shutil.copy2(os.path.join(src, f), os.path.join(dst, f))
        
            # for this is a file, copy all segments
            if os.path.isfile(os.path.join(src, f)):
                for i in range(1, 66000):
                    i = '.' + str(i)
                    fseg = f + i
                    if os.path.exists(os.path.join(src, fseg)):
                        if os.path.exists(os.path.join(dst,fseg)):
                            logger.debug('rm(%s)' % os.path.join(dst,fseg))
                            os.remove(os.path.join(dst,fseg))
                        logger.debug('cp(%s)' % os.path.join(dst,fseg))
                        shutil.copy2(os.path.join(src, fseg),
                                        os.path.join(dst, fseg))
                    else:
                        break
    #------------------------------------------------------------
    def DirIsIncludedFromBackup(self, d):
        if (d != 'base' and d != 'gpmigrator' and
            d != 'mirror_upgrade_state' and
            d.startswith('gp_dump_') == False and
            d.startswith('gp_cdatabase_') == False and
            d.startswith('pg_log') == False ):
            return True
        return False
    
    #------------------------------------------------------------
    def CreateMasterCopy(self):
        self.ForceShutDown()
        
        """
        Copy everthing under the master data, except base and gpmigrator, to gpmigrator/backup
        Also, exclude gp_dump_*, gp_cdatabase_* and pg_log
        """
        dst = os.path.join(self.masterdir, WORKDIR, BACKUPDIR)
        self.RunCmd('rm -rf ' + dst)
        os.mkdir(dst)
        for d in os.listdir(self.masterdir):
            if (self.DirIsIncludedFromBackup(d)):
                self.do_copy(self.masterdir, dst, d)

        """
        Copy everything under catdir, except the backup dir.
        """
        for src in self.mastercatdir:
            dst = os.path.join(src, BACKUPDIR)
            self.RunCmd('rm -rf ' + dst)
            os.mkdir(dst)
            for d in os.listdir(src):
                if (d != BACKUPDIR):
                    self.do_copy(src, dst, d)

    #------------------------------------------------------------
    def DeleteMasterCatalogCopy(self):
        # just print all the cat dirs
        logger.info("Deleting Master Catalog Backup Copy: %s" % self.mastercatdir)
        
        """
        Delete catdir/backup
        """
        for src in self.mastercatdir:
            dst = os.path.join(src, BACKUPDIR)
            self.RunCmd('rm -rf ' + dst)

        return "DeleteMasterDataDirCopy"

    #------------------------------------------------------------
    def RevertMasterCopyAndCatVersion(self):
        self.ForceShutDown()
        
        """
        Delete everything which doesn't exists in the backup dir
        Copy the backup to the cat dir
        """
        for catdir in self.mastercatdir:
            backupdir = os.path.join(catdir, BACKUPDIR)
            for d in os.listdir(catdir):
                if (d != BACKUPDIR):
                    backuploc = os.path.join(backupdir, d)
                    if (os.path.exists(backuploc)):
                        self.do_copy(backupdir, catdir, d)
                    else:
                        dst = os.path.join(catdir, d)
                        self.RunCmd('rm -rf ' + dst)

        """
        Delete everything under the master data, except base and gpmigrator, 
        mirror_upgrade_state file, and pg_log dir
        to gpmigrator/backup
        Copy things from backup
        """
        backupdir = os.path.join(self.masterdir, WORKDIR, BACKUPDIR)
        for d in os.listdir(self.masterdir):
            if (self.DirIsIncludedFromBackup(d)):
                backuploc = os.path.join(self.masterdir, WORKDIR, BACKUPDIR, d)
                if (os.path.exists(backuploc)):
                    self.do_copy(backupdir, self.masterdir, d)
                else:
                    dst = os.path.join(self.masterdir, d)
                    self.RunCmd('rm -rf ' + dst)

        """
        Set the catversion to the old one
        """
        self.SetCatVersion(False)
        
        return "SwitchToMirror"

    #------------------------------------------------------------
    def SwitchToMirror(self):
        # Set the mirror as primary
        self.ForceShutDown()
        self.Startup(self.oldenv, True)
        self.Update("""begin;
            set allow_system_table_mods='dml';
            update gp_segment_configuration set mode='c', role='p' where preferred_role='m';
            update gp_segment_configuration set mode='s', role='m', status='d' where preferred_role='p' and content >=0;
            commit;
        """)
        self.Shutdown()
        
        return "RecoverPrimary"

    #------------------------------------------------------------
    def RecoverPrimary(self):
        self.ForceShutDown()
        self.Startup(self.oldenv)
        logger.info("Running gprecoverseg to recover primary....")
        
        cmd = "$GPHOME/bin/gprecoverseg -a -F -l %s" % self.logdir
        p = subprocess.Popen(cmd, shell = True,
                            close_fds = True,
                            stdout = subprocess.PIPE,
                            stderr = subprocess.PIPE,
                            env=self.oldenv)
        result = p.communicate()
        if p.returncode != 0:
            logger.debug(result[0])
            logger.warn("****************************************************")
            logger.warn("Recovering Primary Failed...!!!!")
            logger.warn("   Please examine the log in %s" % self.logdir)
            logger.warn("   fix the problem and gprecoverseg -F to recover the primary")
            logger.warn("****************************************************")
        
        return "AbortMirrorUpgrade"


    #------------------------------------------------------------
    def DeleteMasterDataDirCopy(self):
        # just print all the cat dirs and data dir
        logger.info("Deleting Master Data Dir Backup Copy: %s" % self.masterdir)
        
        """
        Delete the work dir
        """
        dst = os.path.join(self.masterdir, WORKDIR)
        self.RunCmd('rm -rf ' + dst)

        return "FinishMirrorUpgrade"
        
    #------------------------------------------------------------
    def TurnOffMirror(self):
        # Check that mirror and primary are in sync again..!!!
        # Set the mirror off and put the primary on changetracking
        self.ForceShutDown()
        self.Startup(self.oldenv, True)
        
        nonsynccnt = self.Select('''select count(*)
             from pg_catalog.gp_segment_configuration
             where status != 'u' or mode != 's' or role != preferred_role
            ''' )
        if (nonsynccnt[0] > 0):
            raise UpgradeError("Mirror or Segments are down or out of sync")

        self.Update("""begin;
            set allow_system_table_mods='dml';
            update gp_segment_configuration set mode='c' where preferred_role='p' and content >=0;
            update gp_segment_configuration set status='d' where preferred_role='m';
            commit;
        """)
        self.Shutdown()
        return "CatUpgrade"

    #------------------------------------------------------------
    def CatUpgrade(self):
        newversion = GpVersion(self.getversion(self.newhome, self.newenv))
        snv        = newversion.getVersionRelease()
        
        self.ForceShutDown()
        
        try:
            # Upgrade pg_control first, and start up the database with
            # the upgrade flag on.
            self.SetCatVersion(True)
            self.Startup(self.newenv, False, True)
            runfile = "%s/share/postgresql/upgrade/upg2_catupgrade.sql" % self.newhome
            self.ExecuteSQLFileOnAllDBs(runfile)
            self.Shutdown()
            self.Startup(self.newenv, False, False)
            self.PerformPostUpgrade()
        except:
            logger.error(traceback.format_exc())
            return "RevertMasterCopyAndCatVersion"
        
        return "RecoverMirror"

    #------------------------------------------------------------
    def RecoverMirror(self):
        self.ForceShutDown()
        self.Startup(self.newenv)
        logger.info("Running gprecoverseg to recover mirror....")

        cmd = ' '.join(["$GPHOME/bin/gprecoverseg", '-a', '-l', self.logdir])
        p = subprocess.Popen(cmd, shell = True,
                            close_fds = True,
                            stdout = subprocess.PIPE,
                            stderr = subprocess.PIPE,
                            env=self.newenv)
        result = p.communicate()
        
        if p.returncode != 0:
            logger.debug(result[0])
            logger.warn("****************************************************")
            logger.warn("Recovering Mirror Failed...!!!!")
            logger.warn("   Please examine the log in %s" % self.logdir)
            logger.warn("   fix the problem and re-run gpmigrator_mirror to recover the mirror")
            logger.warn("****************************************************")

        return "DeleteMasterCatalogCopy"
    
    #------------------------------------------------------------
    def LoadState(self):
        # Setup the current state and statefile
        #   if state file exists, the current state is the last state
        self.mirror_upgrade_state = "CreateMirrorWorkDir"
        statefile                 = os.path.join(self.masterdir, 'mirror_upgrade_state')
        hasconfigfile             = False
        if os.path.exists(statefile):
            logger.info("Found previous upgrade")
            self.mirror_upgrade_state_file = open(statefile, 'r')
            for line in self.mirror_upgrade_state_file:
                line = line.strip()
                if (line == "TurnOffMirror"):
                    hasconfigfile = True
                if (line == 'DeleteMasterDataDirCopy'):
                    hasconfigfile = False
                self.mirror_upgrade_state = line
            self.mirror_upgrade_state_file.close()
        self.mirror_upgrade_state_file = open(statefile, 'a')
        
        if (hasconfigfile):
            self.ReadInfo()

    #------------------------------------------------------------
    def AbortMirrorUpgrade(self):
        self.mirror_upgrade_state_file.close()
        statefile = os.path.join(self.masterdir, 'mirror_upgrade_state')
        open(statefile, 'w').truncate()

        logger.info("---------------------------------------------------")
        logger.info("Upgrade Aborted")
        logger.info("---------------------------------------------------")
        logger.info("Please fix up the error and re-run upgrade")
        logger.info("---------------------------------------------------")

        sys.exit(0)
    #------------------------------------------------------------
    def FinishMirrorUpgrade(self):
        self.mirror_upgrade_state_file.close()
        statefile = os.path.join(self.masterdir, 'mirror_upgrade_state')
        self.RunCmd('rm -rf ' + statefile)
        
        logger.info("---------------------------------------------------")
        logger.info("Upgrade Successful")
        logger.info("---------------------------------------------------")
        logger.info("Please consult release notes for post-upgrade ")
        logger.info("instructions to complete environment configuration")
        logger.info("---------------------------------------------------")

        sys.exit(0)

    #------------------------------------------------------------
    def PerformMirroredUpgrade(self):
        #==========================================================
        # Upgrade strategy
        #   Use mirror as backup: start the db with mirror off and put primary in changetracking.
        #   Then start the db in upgrade mode to run the upgrade script as usual.
        #   When the script finished, use gprecoverseg to bring the mirror back in sync.
        #   In case of error when running the upgrade script, set the primary as down and bring
        #   the mirror up as primary. Do either full recovery (gprecoverseg -F) or delta
        #   (gpcheckmirrorseg, gprepairmirrorseg and gprecoverseg).
        #
        #   Master is not mirrored. We'll copyed everything in the master to a backup location
        #==========================================================

        # We use a state machine approach to handle fault and crash.
        # Fault will be cought as exception and it'll go the fault hanlding
        # state.
        # Crash will not change the state. When restart, it'll re-do the current state
        # agagin.

        # Initialize the current state 
        #   if a state file already exists, the current state is in the state file
        self.LoadState()
        
        faultinject_index = 100
        
        while True:
            logger.info("  ")
            logger.info("**************************************************")
            logger.info("**************************************************")
            logger.info("Executing state:%s"%self.mirror_upgrade_state)
            logger.info("**************************************************")
            logger.info("**************************************************")
            if (self.mirror_upgrade_state == "CreateMirrorWorkDir"):
                self.mirror_upgrade_state = self.CreateMirrorWorkDir()
            elif (self.mirror_upgrade_state == "ExtractInfoAndCopyMaster"):
                self.mirror_upgrade_state = self.ExtractInfoAndCopyMaster()
            elif (self.mirror_upgrade_state == "TurnOffMirror"):
                self.mirror_upgrade_state = self.TurnOffMirror()
            elif (self.mirror_upgrade_state == "CatUpgrade"):
                self.mirror_upgrade_state = self.CatUpgrade()
            elif (self.mirror_upgrade_state == "RecoverMirror"):
                self.mirror_upgrade_state = self.RecoverMirror()
            elif (self.mirror_upgrade_state == "DeleteMasterCatalogCopy"):
                self.mirror_upgrade_state = self.DeleteMasterCatalogCopy()
            elif (self.mirror_upgrade_state == "DeleteMasterDataDirCopy"):
                self.mirror_upgrade_state = self.DeleteMasterDataDirCopy()
            elif (self.mirror_upgrade_state == "RevertMasterCopyAndCatVersion"):
                self.mirror_upgrade_state = self.RevertMasterCopyAndCatVersion()
            elif (self.mirror_upgrade_state == "SwitchToMirror"):
                self.mirror_upgrade_state = self.SwitchToMirror()
            elif (self.mirror_upgrade_state == "RecoverPrimary"):
                self.mirror_upgrade_state = self.RecoverPrimary()
            elif (self.mirror_upgrade_state == "AbortMirrorUpgrade"):
                self.mirror_upgrade_state = self.AbortMirrorUpgrade()
            elif (self.mirror_upgrade_state == "FinishMirrorUpgrade"):
                self.mirror_upgrade_state = self.FinishMirrorUpgrade()
            else:
                # can't get here
                raise UpgradeError('Unknown Command : %s' % self.mirror_upgrade_state)
                sys.exit(1)
                
            if self.faultinjection == faultinject_index:
                    raise UpgradeError("faultinjection=%d" % self.faultinjection)
            faultinject_index = faultinject_index+1

            self.mirror_upgrade_state_file.write("%s\n" % self.mirror_upgrade_state)
            self.mirror_upgrade_state_file.flush()
        sys.exit(0)


#============================================================
if __name__ == '__main__':
    coverage = GpCoverage()
    coverage.start()
    
    # Create a new GPUpgrade - should never throw exception
    u = GPUpgrade()
    
    # parses and validates input
    try:
        u.Setup()
    except Exception, e:
        logger.fatal(str(e))
        sys.exit(1)
    
    # Execute the main upgrade routine, the Run() function itself is
    # just a switch on the "cmd", which switches between the main
    # PerformUpgrade() function and all the various subcommands that
    # are upgraded by backend executions.
    try:
        u.Run()
    # Any error that can happen
    except KeyboardInterrupt:
        logger.fatal('***************************************')
        logger.fatal('==  Upgrade interrupted by user ==        <<<<< ')
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    except ConnectionError, e:
        logger.fatal('***************************************')
        logger.fatal(str(e))
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    except UpgradeError, e:
        logger.fatal('***************************************')
        logger.fatal(str(e))
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    except CmdError, e:
        logger.fatal('***************************************')
        logger.fatal('Error executing command:')
        logger.fatal('  ' + str(e.cmd.strip()))
        logger.fatal('  ' + str(e.stderr.strip()))
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    # Shouldn't get any of these, if they occur it is probably a bug in
    # the upgrader.
    except Exception, e:
        logger.fatal('***************************************')
        logger.fatal(str(e))
        logger.fatal('***************************************')
        logger.fatal(traceback.format_exc())
        sys.exit(1)
    finally:
        coverage.stop()
        coverage.generate_report()

    sys.exit(0)

916fd3510907dff865ef5c60cbffc67fbc9f2e87
#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
gpmigrator_mirror [options] old-gphome new-gphome

Options:
  -h, --help            show this help message and exit
  -v, --version         show the program's version number and exit
  -q                    quiet mode
  -d DIRECTORY          the master host data directory
  -l DIRECTORY          log file directory
  --debug               debugging information
'''

#============================================================
import sys, os

# Python version 2.6.2 is expected, must be between 2.5-3.0
if sys.version_info < (2, 5, 0) or sys.version_info >= (3, 0, 0):
    sys.stderr.write("Error: %s is supported on Python versions 2.5 or greater\n" 
                     "Please upgrade python installed on this machine." 
                     % os.path.split(__file__)[-1])
    sys.exit(1)

try:
    from gppylib.operations.gpMigratorUtil import *
except ImportError, e:
    sys.exit('Error: unable to import module: ' + str(e))

libdir = os.path.join(sys.path[0], 'lib/')

logger        = get_default_logger()
EXECNAME      = os.path.split(__file__)[-1]
MIGRATIONUSER = 'gpmigrator'
LOCKEXT       = '.gpmigrator_orig'
WORKDIR       = 'gpmigrator'
BACKUPDIR     = 'backup'
UPGRADEDIR    = 'upgrade'
PARALLELISM   = 16




#============================================================
__version__ = '$Revision: #1 $'




#============================================================
def makeCommand(oldHome, newHome,
                 command, dataDirectories, option1):
    # space separated list of directories
    datadirs = base64.urlsafe_b64encode(pickle.dumps(dataDirectories))
    cmd = [
           EXECNAME,
           oldHome, 
           newHome,
           '--internal-command=' + str(command),
           '--internal-dirs=' + urllib.quote(datadirs),
           '--internal-option=' + urllib.quote(str(option1)),
           '--debug',
           '--quiet',
           ]
    return cmd

#============================================================
def logGenericErrorMessage():
    logger.fatal('Upgrade Failed..!!!')
    logger.fatal('...Database might be in an unusable state')
    logger.fatal('...Please check the logs, resolve the issue an re-run gpmigrator_mirror')


#============================================================
class GPUpgrade(GPUpgradeBase):
    '''
    Greenplum Database Upgrade Utility
    '''

    # INTERNAL command:
    #   MASTER    - default, run on master, [not included in list]
    #   CHKDOWN   - check the db is down
    #   SETCATVERSION - set catalog version
    commands = ['CHKDOWN', 'SETCATVERSION']

    #------------------------------------------------------------
    def __init__(self):
        '''
        The most basic of initialization
        '''
        super(GPUpgrade, self).__init__()
        
        # Environment and system info
        self.datadirs  = None     # Set of all data and mirror directories (Segment only)

        
        # Mirrored Upgrade State
        self.mirror_upgrade_state = None
        self.mirror_upgrade_state_file = None

    #------------------------------------------------------------
    def Setup(self):
        '''
        Basic initialization, separate from __init__ for exception
        handling purposes.
        '''

        # GPHOME, PYTHONPATH must be setup properly
        #    GPHOME must match the location of this file
        #    The first PYTHONPATH must be based on the proper gphome 
        gphome_bin = os.path.realpath(os.path.split(__file__)[0])
        gphome     = os.path.split(gphome_bin)[0]
        env_GPHOME = os.path.realpath(os.environ.get('GPHOME'))
        if (env_GPHOME != gphome):
            logger.fatal(" $GPHOME is set to %s which is not newhome" % env_GPHOME)
            logger.fatal(' source the greenplum.sh from the newhome to setup env ')
            raise UpgradeError('Initialization failed')

        pythonpath     = os.path.join(gphome, "lib", "python")
        env_PYTHONPATH = os.path.realpath(os.environ.get('PYTHONPATH').split(':')[0])
        if (env_PYTHONPATH != pythonpath):
            logger.fatal(' $PYTHONPATH is incorrect ')
            logger.fatal(' source the greenplum.sh from the newhome to setup env ')
            raise UpgradeError('Initialization failed')

        # This is the same path used by gpinitsystem
        self.path = '/usr/kerberos/bin:/usr/sfw/bin:/opt/sfw/bin'
        self.path += ':/usr/local/bin:/bin:/usr/bin:/sbin:/usr/sbin:/usr/ucb'
        self.path += ':/sw/bin'

        # Set defaults
        self.user      = os.environ.get('USER') or os.environ.get('LOGNAME')
        self.host      = self.RunCmd('hostname')
        self.masterdir = os.environ.get('MASTER_DATA_DIRECTORY')
        
        self.ParseInput()
        
        # Setup worker pool
        self.pool = base.WorkerPool(numWorkers=PARALLELISM);

        # Extract path from input master directory
        if self.cmd == 'MASTER':
            logger.info('Beginning upgrade')
            logger.info('Checking configuration')

            if not self.masterdir or len(self.masterdir) == 0:
                raise UpgradeError('MASTER_DATA_DIRECTORY is not defined')
            self.masterdir = self.masterdir.rstrip('/')

            # The configuration file
            conf = os.path.join(self.masterdir, 'postgresql.conf')

            # Simple function to look for settings in the conf file
            def getconf(x):
                conf_re  = re.compile('^\s*%s\s*=\s*(\w+)' % x)
                try:
                    conf_str = self.RunCmd('grep %s %s' % (x, conf))
                except CmdError:
                    conf_str = ""     # grep returns errorcode on no match
                value = None
                for line in conf_str.split('\n'):
                    match = conf_re.search(line)
                    if match:
                        value = match.group(1)
                return value

            # Find the port for this segment:
            self.masterport = getconf('port')
            if self.masterport == None:
                raise UpgradeError('Could not determine master port from ' + conf)
            self.masterport = int(self.masterport)

            self.sock_dir = getconf('unix_socket_directory')
            if not self.sock_dir:
                self.sock_dir = '/tmp/'

            # Verify that (max_connections == superuser_reserved_connections)
            # max_conn = getconf('max_connections')
            # reserved = getconf('superuser_reserved_connections')

            masterpath, masterdir = os.path.split(self.masterdir)
            self.workdir   = os.path.join(self.masterdir, WORKDIR)
            self.oldenv    = self.SetupEnv(self.oldhome, self.masterdir)
            self.newenv    = self.SetupEnv(self.newhome, self.masterdir)

        else:
            self.oldenv = self.SetupEnv(self.oldhome, None)
            self.newenv = self.SetupEnv(self.newhome, None)

    #------------------------------------------------------------
    def ParseInput(self):
        '''
        Parses and validates input to the script
        '''

        try:
            parser = optparse.OptionParser(usage=(cli_help(EXECNAME) or __doc__), add_help_option=False)
            parser.add_option('-v', '--version', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-h', '--help', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-d', dest='directory', 
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-l', dest='logdir',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-c', '--check-only', dest='checkonly', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('-q', '--quiet', action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--debug', action='store_true',
                              help=optparse.SUPPRESS_HELP)

            parser.add_option('--internal-command', dest='command',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--internal-dirs', dest='datadirs',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--internal-option', dest='option',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--skip-checkcat', dest='skipcheckcat',
                              action='store_true',
                              help=optparse.SUPPRESS_HELP)
            parser.add_option('--fault-injection', dest='faultinjection',
                              type='int', help=optparse.SUPPRESS_HELP)
            (options, args) = parser.parse_args()
            
            if options.version:
                print EXECNAME + ' ' + __version__
                sys.exit(0)

            if options.help:
                usage(EXECNAME)
                sys.exit(0)

            if len(args) != 2:
                usage(EXECNAME)
                msg = "incorrect number of arguments"
                if len(args) > 0:
                    msg += ": %s" % str(args)
                parser.error(msg)
        except Exception, e:
            usage(EXECNAME)
            raise UpgradeError('Error parsing input: ' + str(e))

        if options.directory:
            self.masterdir = options.directory
        if options.command:
            self.cmd = options.command
        self.oldhome   = args[0].rstrip('/')
        self.newhome   = args[1].rstrip('/')
        if options.datadirs:
            self.datadirs = pickle.loads(base64.urlsafe_b64decode((urllib.unquote(options.datadirs))))
        if options.option:
            self.option   = urllib.unquote(options.option)
        if options.checkonly:
            self.checkonly = True
        if options.debug:
            self.debug = True
            enable_verbose_logging()
        if options.quiet:
            quiet_stdout_logging()
        if options.logdir:
            self.logdir = options.logdir
        else:
            self.logdir = os.path.join(os.environ['HOME'], 'gpAdminLogs')

        try:
            setup_tool_logging(EXECNAME,
                               unix.getLocalHostname(),
                               unix.getUserName(),
                               self.logdir)
        except OSError, e:
            logger.fatal('cannot log to %s: %s' % (self.logdir, str(e)))
            exit(1)
        
        self.skipcheckcat = options.skipcheckcat
        self.faultinjection = options.faultinjection

        if self.cmd == 'MASTER':

            today = date.today().strftime('%Y%m%d')
            if not os.path.isdir(self.logdir):
                os.makedirs(self.logdir, 0700)
            logname = os.path.join(self.logdir, '%s_%s.log' % (EXECNAME, today))
            self.logfile = open(logname, 'a')

            if self.masterdir == None:
                logger.fatal('MASTER_DATA_DIRECTORY parameter not set')
                raise UpgradeError('Initialization failed')
            if not os.path.exists(self.masterdir):
                logger.fatal('MASTER_DATA_DIRECTORY: %s not found'
                             % self.masterdir)
                raise UpgradeError('Initialization failed')
            if not os.path.isabs(self.masterdir):
                self.masterdir = os.path.abspath(self.masterdir)
            if not os.path.exists(self.oldhome):
                logger.fatal(' directory not found: ' + self.oldhome)
                raise UpgradeError('Initialization failed')
            if not os.path.exists(self.newhome):
                logger.fatal(' directory not found: ' + self.newhome)
                raise UpgradeError('Initialization failed')
            if not os.path.isabs(self.oldhome):
                self.oldhome = os.path.abspath(self.oldhome)
            if not os.path.isabs(self.newhome):
                self.newhome = os.path.abspath(self.newhome)


    #------------------------------------------------------------
    def CallSlaves(self, cmd, option=''):
        '''
        Calls every host to execute the given command with the given option 
        value
        '''
  
        logger.debug("Remote call: %s" % cmd)

        # Check for things that should never happen
        if self.cmd != 'MASTER':
            raise UpgradeError("Recursive communication error")
        if not self.array:
            raise UpgradeError("Failure initializing array")
        if not self.hostcache:
            raise UpgradeError("Failure initializing host cache")
        if not self.pool:
            raise UpgradeError("Failure initializing worker pool")
        
        # Construct the commands to pass to the worker pool
        hosts = self.hostcache.get_hosts()
        for host in hosts:
            hostname = host.hostname

            # Skip any hosts in the cache that contain no segments for this
            # configuration.
            if len(host.dbs) == 0:
                continue

            # Get the data directories for this host:
            datadirs = []
            for seg in host.dbs:
                datadirs.append(seg.getSegmentDataDirectory())
            
            # Skip any hosts that have no applicable data directories
            if len(datadirs) == 0:
               continue

            cmdList = makeCommand(oldHome=self.oldhome,
                             newHome=self.newhome,
                             command=cmd,
                             dataDirectories=datadirs,
                             option1=option)

            c = GpUpgradeCmd("gpmigrator_mirror remote call",
                             cmdList,
                             ctxt=base.REMOTE,
                             remoteHost=hostname)
            self.pool.addCommand(c)

        # Wait for the segments to finish
        try:
            self.pool.join()
        except:
            self.pool.haltWork()
            self.pool.joinWorkers()
            
        failure = False
        results = []
        for cmd in self.pool.getCompletedItems():
            r = cmd.get_results()

            # Going through the gppylib Command interface all stderr from the
            # remote calls gets redirected to stdout, which is unfortunate
            # because we'd like to be able to differentiate between the two.
            #
            # We keep the stdout chatter to a minimum by passing --quiet to
            # the remote calls which performs quiet stdout logging.

            # sys.stderr.write(r.stderr)
            msg = r.stdout.strip()
            results.append(msg)
            
            if not cmd.was_successful():
                log_literal(logger, logging.ERROR, msg)
                failure=True

        if failure:
            raise UpgradeError("Fatal Segment Error")

        # Warning this output contains everything written to stdout,
        # which unfortunately includes some of the logging information
        return "\n".join(results)


    #------------------------------------------------------------
    def CheckVersions(self):
        '''
        Validates that the upgrade from old->new is okay
        '''

        # Log the OS type info
        os_type = os.uname()[0]
        os_ver  = os.uname()[2]
        logger.info('Operating System: %s %s' % (os_type, os_ver))

        # Log version checking
        logger.info('Checking version compatibility')

        # If we got a version, but it isn't recognized by the GpVersion class
        # then it is likely not a Greenplum database.
        oldversion = self.getversion(self.oldhome, self.oldenv)
        newversion = self.getversion(self.newhome, self.newenv)
        try:
            oldversion = GpVersion(oldversion)
        except:
            raise UpgradeError('Source not a Greenplum Database: ' + oldversion)
        try:
            newversion = GpVersion(newversion)
        except:
            raise UpgradeError('Target not a Greenplum Database: ' + newversion)

        logger.info('Source Version: (Greenplum Database) %s' % str(oldversion))
        logger.info('Target Version: (Greenplum Database) %s' % str(newversion))

        if newversion == oldversion:
            raise UpgradeError("Greenplum Database is already version '%s'" 
                               % str(newversion))
        elif newversion.isVersionRelease(oldversion):
            raise UpgradeError("Upgrade not needed to go from version '%s' to version '%s'" 
                               % (str(oldversion), str(newversion)))

        is_supported_version(oldversion)
        is_supported_version(newversion)

        # We don't support downgrade
        if newversion < oldversion:
            raise UpgradeError("Downgrade to Greenplum Database %s not supported from gpmigrator" 
                               % str(newversion))

        if not newversion.isVersionCurrentRelease():
            main = GpVersion('main')
            raise UpgradeError(
                "Upgrade from '%s' to '%s' not supported.  Target version should be Greenplum Database %s"
                % (str(oldversion), str(newversion), main.getVersionRelease()))

        # Compare old version catalog number with the catalog in 
        # MASTER_DATA_DIRECTORY
        catalog_re = re.compile('Catalog version number: *(.*)')
        oldcat = self.RunCmd('postgres --catalog-version', env=self.oldenv)
        logger.info("Source %s" % oldcat)
        newcat = self.RunCmd('postgres --catalog-version', env=self.newenv)
        logger.info("Target %s" % newcat)

        control = self.RunCmd('pg_controldata ' + self.masterdir, env=self.oldenv)
        for line in control.split('\n'):
            m = catalog_re.match(line)
            if m:
                logger.info("Data   %s " % line)
                if (line != oldcat):
                    logger.debug('catalog mismatch: expected %s, found %s'
                             % (oldcat, line))
                    msg = 'Catalog in %s does not match source binary' % self.masterdir
                    raise UpgradeError(msg)
                break

        # For the moment everything goes:
        #   - Additional checks will be made once the old database is up
        logger.info('Versions are compatible')        
               

    #------------------------------------------------------------
    def GenerateConfigFiles(self):
        '''
        Creates 'gp_databases' 'gp_array_config' and 'gp_config' files
        '''

        logger.info('Generating config files')        

        # Write a dump of the databases hash
        db_file = os.path.join(self.workdir, 'gp_databases')
        o = open(db_file, 'w')
        for (oid, name) in self.dbs.iteritems():
            o.write('%s:%s\n' % (oid, name))
        o.close()

        # Write a dump of the gparray object
        array_file = os.path.join(self.workdir, 'gp_array_config')
        self.array.dumpToFile(array_file)

        logger.info('Configuration files generated')


    #------------------------------------------------------------
    def ReadInfo(self):
        ''' 
        When restarting an upgrade this method reads the configuration
        information from the cached status files.  

        It corresponds directly with ExtractInfo() which gathers the same
        information from an active database.

        It is also the inverse of GenerateConfigFiles() which creates the files
        that this function reads.

        In normal processing usually ExtractInfo() is called rather than
        ReadInfo().  This function is used /instead/ when we are doing Revert()
        processing, or when resuming an upgrade.  E.g. cases when we do not
        want to start the database to get the information we need because the
        database may not be in a stable state.
        '''

        # self.array - configuration information
        array_config = os.path.join(self.workdir, 'gp_array_config')
        if not os.path.exists(array_config):
            raise UpgradeError("  '%s' not found" % array_config)
        self.array = GpArray.initFromFile(array_config)

        # self.dbs - database information
        db_file = os.path.join(self.workdir, 'gp_databases')
        if not os.path.exists(db_file):
            raise UpgradeError("  '%s' not found" % db_file)
        f = open(db_file, 'r')
        for line in f:
            (oid, name) = line.split(':', 1)
            self.dbs[oid] = name.strip()
        f.close()

        # self.hostcache - cache of hostnames
        self.hostcache = GpHostCache(self.array, self.pool)
        failed_pings = self.hostcache.ping_hosts(self.pool)
        if len(failed_pings) > 0:
            raise UpgradeError(
                "Cannot upgrade while there are unreachable hosts")

        # self.mirrors
        dbs = self.array.getSegDbList()
        self.mirrors  = filter(GpDB.isSegmentMirror, dbs)
                
        # Setup the master catalog dirs
        self.mastercatdir = self.array.master.catdirs



    #------------------------------------------------------------
    def ExtractInfo(self):
        '''
        Get all the information we need from the old instance
        '''

        # Can only extract with a running database
        env = self.CheckUp()

        # -- Get the array information
        logger.info('Extracting configuration information')
        port  = self.masterport
        user  = env['USER']
        url   = dbconn.DbURL(port=port, dbname='template1', username=user)
        array = GpArray.initFromCatalog(url, utility=True)
        self.array = array

        # -- Determine if there are any invalid segments
        logger.info('Checking that all segments are valid')
        invalid = array.get_invalid_segdbs()
        if len(invalid) > 0:
            for db in invalid:
                logger.fatal('INVALID SEGMENT: %s:/%s' % 
                             (db.getSegmentHostName(),
                              db.getSegmentDataDirectory()))
            raise UpgradeError('Cannot upgrade database with invalid segments')


        # -- Get a list of available databases, note this includes "template0"!
        logger.info('Creating list of databases')
        databases = self.Select("SELECT oid, datname from pg_database")
        for row in databases:
            self.dbs[str(row['oid'])] = row['datname']

        # -- Older releases didn't have hostname as part of configuration
        # make sure we have an acurate list of address->host lookups.
        logger.info('Validating hosts')
        self.hostcache = GpHostCache(self.array, self.pool)
        for host in self.hostcache.get_hosts():
            for seg in host.dbs:
                seg.setSegmentHostName(host.hostname)
        failed_pings = self.hostcache.ping_hosts(self.pool)
        if len(failed_pings) > 0:
            raise UpgradeError("Cannot upgrade while there are unreachable "
                               "hosts")
            
        # Setup the master catalog dirs
        self.mastercatdir = self.array.master.catdirs
            
        master = array.master
        standbymaster = array.standbyMaster

        # Oddly gparray doesn't have methods for fetching primary segments
        # specifically
        dbs = array.getSegDbList()
        segments = filter(GpDB.isSegmentPrimary, dbs)
        self.mirrors  = filter(GpDB.isSegmentMirror, dbs)

        # Internal sanity checking
        if len(segments) == 0:
            raise UpgradeError('No segments found')
        if standbymaster > 0:
            raise UpgradeError('Cannot upgrade while standbymaster is running')

        # Check for unsupported index types.
        found = False
        logger.info("Validating indexes")
        oids = sorted(self.dbs.keys())
        for dboid in oids:
            db = self.dbs[dboid]

            if db == "template0": 
                continue
            indexes = self.Select('''
               SELECT quote_ident(n.nspname) || '.' || 
                      quote_ident(c.relname) as index, 
                      m.amname as kind
               FROM   pg_class c 
                 join pg_namespace n on (c.relnamespace = n.oid) 
                 join pg_am m on (c.relam = m.oid) 
               WHERE  c.relkind = 'i' and m.amname in ('gin', 'hash')
            ''', db=db)
            if len(indexes) > 0:
                if found == False:
                    logger.fatal("Unable to upgrade the following indexes")
                    found = True
                logger.fatal("  Database: %s" % db)
                for row in indexes:
                    logger.fatal("     %s  [%s]" % (row['index'], row['kind']))
        if found:
            raise UpgradeError("Deprecated index types must be removed prior "
                               "to upgrade")

        # Check for SQL_ASCII database encoding
        logger.info("Validating database encoding")
        encoding = self.Select("SELECT datname FROM pg_database WHERE encoding=0")
        if len(encoding) > 0:
            logger.error("Deprecated database encodings found:")
            for datname in encoding:
                logger.error("    %s  [SQL_ASCII]" % datname)
            raise UpgradeError("Deprecated database encodings found - contact "
                               "Greenplum Support")

        logger.info('Configuration acquired')
                    
    #------------------------------------------------------------
    def ExecuteSQLFileOnAllDBs(self, runfile):
        '''
        Execute "runfile" in upgrade mode on all databases, including template0
        '''
        def transform_db(dbname, dboid, tfile):
            '''
            Run the script on the specified database
            '''
            logger.debug("Executing on db = %s dboid = %s tfile = %s" % \
                     (str(dbname), str(dboid), tfile))

            # Execute the upgrade script
            cmd = ' '.join(["PGOPTIONS='-c gp_maintenance_conn=true'", 'psql', '-f', tfile, dbname, self.user])
            logger.debug('running ' + cmd)
            p = subprocess.Popen(cmd, shell = True,
                                  close_fds = True,
                                  stdout=subprocess.PIPE, 
                                  stderr=subprocess.PIPE, 
                                  env=self.newenv)
            result = p.communicate()
        
            if p.returncode == 0:
                logger.debug('successfully transformed ' + dbname)
                logger.debug(result[0].strip())
                logger.debug(result[1].strip())
            else:
                logger.debug(result[0])
                raise CmdError(str(cmd), result[0], result[1])


        logger.info('Performing catalog transformation')
        # MPP-10166 - because some of our transformations involve creates
        # of views with unstable oids we must make sure that we process 
        # the databases in exactly the same order on every segment, so
        # sorting the oids is essential.
        oids = sorted(self.dbs.keys())
        for dboid in oids:
            db = self.dbs[dboid]
                
            logger.info('... ' + db)
            transform_db(db, dboid, runfile)
            # Test failure during transformation
            if self.faultinjection == 1:
                raise UpgradeError("faultinjection=%d" % self.faultinjection)

    #------------------------------------------------------------
    def SetCatVersion(self, tonewver):
        '''
        Upgrade/downgrade pg_control file to another version. Cluster should
        be clean shutdown.
        '''

        logger.info('Modifying catalog version')
        
        if (tonewver==True):
            curenv     = self.newenv
            curhome    = self.newhome
        else:
            curenv     = self.oldenv
            curhome    = self.oldhome
        toversion = GpVersion(self.getversion(curhome, curenv))
        to        = toversion.getVersionRelease()

        logger.info("To %s"%to)
        self.CheckDown()

        def get_control_data(datadir):
             '''
             Parse the output of pg_controldata run on data directory, returning
             catalog version and state
             '''
             cmd = ' '.join(['pg_controldata', datadir])
             p = subprocess.Popen(cmd, shell=True, close_fds=True,
                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                  env=curenv)
             result = p.communicate()
        
             if p.returncode != 0:
                 raise CmdError(cmd, result[0], result[1])
        
             out = result[0].strip()
             ver = ""
             state = ""
             for line in out.split('\n'):
                  s = line.split(':')
                  if s[0] == 'Catalog version number':
                        ver = s[1].strip()
                  elif s[0] == 'Database cluster state':
                        state = s[1].strip()
        
             return [ver, state]
        
        def setcatversion(datadir, to):
            '''
            Set catalog version to 'to'

            We use gpmodcontrol to update pg_control file since in this
            version (4.3) we change the ControlFile structure.  If only
            catalog_version_no needs to change, we could use gpmodcatversion
            as was previously.  In case of gpmodcontrol (but not the case
            for gpmodcatversion), the newer environment is responsible to
            take care of downgrade too since older version does not have
            knowledge of the newer structure.
            '''
            # It is an error if the current state is not "shut down", that
            # means that there may be pending xlog records which will cause 
            # problems for the upgrade.
            (ver, state) = get_control_data(datadir)
            if state != "shut down":
                raise UpgradeError("Cannot upgrade: Database did not shutdown cleanly")

            cmd = ['%s/bin/lib/gpmodcontrol' % self.newhome]
            if not tonewver:
                cmd.append('--downgrade')
            cmd.append(datadir)
            p = subprocess.Popen(cmd,
                                 shell=False,
                                 close_fds=True,
                                 env=self.newenv, # always newenv
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)
            result = p.communicate()
            if p.returncode != 0:
                raise Exception("could not update catalog to %s" % to)


        if self.cmd == 'MASTER':
            datadirs = [ self.masterdir ]
            self.CallSlaves('SETCATVERSION', tonewver)
            if self.faultinjection == 2:
                raise UpgradeError("faultinjection=%d" % self.faultinjection)
        else:
            datadirs = self.datadirs

        for dir in datadirs:
            setcatversion(dir, to)
            if self.faultinjection == 3:
                raise UpgradeError("faultinjection=%d" % self.faultinjection)
            

        
    #------------------------------------------------------------
    def Run(self):
        """
        The main execution method - switches between submodes.

        The main frontend will execute the main PerformUpgrade() function,
        during which it will execute CallSlaves with various backend operations.
        For each one of the main backend operations we will spawn the backend
        executables, this switches between each of these operations.
        """

        # Execution on the master
        if self.cmd == 'MASTER':
            self.PerformMirroredUpgrade()
        
        # The rest of the options are executed on remote hosts
        elif self.cmd == 'CHKDOWN':
            self.CheckDown()
        elif self.cmd == 'SETCATVERSION':
            self.SetCatVersion((str(self.option)==str(True)))
        else:
            raise Exception('Unknown cmd: ' + str(self.cmd))
    
        if self.pool:
            t = self.pool
            self.pool = None
            del t

    # MIRROR UPGRADE CODE
    #------------------------------------------------------------
    def ForceShutDown(self):
        # The cleanest shutdown is to start the master in master only
        # and then do gpstop -a.
        # This simple thing couldn't shutdown segment if the master is dead.
        try:
            env = self.CheckUp()
            self.dbup = [env, False]
            self.Shutdown()
        except:
            pass


    def CreateMirrorWorkDir(self):
        # Create the workdir
        #   if the dir exists, clean it first
        self.RunCmd('rm -rf ' + self.workdir)
        os.mkdir(self.workdir)
        return "ExtractInfoAndCopyMaster"

    #------------------------------------------------------------
    def ExtractInfoAndCopyMaster(self):
        """
        Extract the database info and ceck for upgrade pre-requisite
        Serialize the db info to a flat file for "resume" (if crashed)
        Create a Master Copy
        """
        self.ExtractandWriteInfo()
        self.CreateMasterCopy()
        return "TurnOffMirror"
    
    def ExtractandWriteInfo(self):
        self.CheckVersions()
        self.ForceShutDown()
        self.Startup(self.oldenv)

        # Pre-requisite check
        #   1. all segments must be up
        #.  2. all mirror and segments must be in sync
        #   3. all mirror/segment must be in their preferred role
        #   4. it must have mirror
        #   5. pass gpcheckcat
        
        # This query check 1, 2 and 3.
        nonsynccnt = self.Select('''select count(*)
             from pg_catalog.gp_segment_configuration
             where status != 'u' or mode != 's' or role != preferred_role
            ''' )
        if (nonsynccnt[0] > 0):
            raise UpgradeError("Mirror or Segments are down or out of sync")

        # Now, extract the info before going on with more checks
        self.ExtractInfo()
        
        #   4. it must have mirror
        if (len(self.mirrors) ==0):
            raise UpgradeError("This upgrade requires mirroring.")
        
        #   5. pass common pre-upgrade check
        self.PreUpgradeCheck()
        
        # If we're in check only mode, then we exit here
        if self.checkonly:
            sys.exit(0)
        
        # Everything is ok. Write info to disk
        self.GenerateConfigFiles()

    #------------------------------------------------------------
    def do_copy(self, src, dst, f):
        '''
        Copy a directory tree
        '''
        if os.path.isdir(os.path.join(src, f)):
            if os.path.exists(os.path.join(dst,f)):
                logger.debug('rmtree(%s)' % os.path.join(dst,f))
                shutil.rmtree(os.path.join(dst,f))
            logger.debug('cptree(%s)' % os.path.join(dst,f))
            shutil.copytree(os.path.join(src, f), os.path.join(dst, f))
        else:
            if os.path.exists(os.path.join(dst,f)):
                logger.debug('rm(%s)' % os.path.join(dst,f))
                os.remove(os.path.join(dst,f))
            logger.debug('cp(%s)' % os.path.join(dst,f))
            shutil.copy2(os.path.join(src, f), os.path.join(dst, f))
        
            # for this is a file, copy all segments
            if os.path.isfile(os.path.join(src, f)):
                for i in range(1, 66000):
                    i = '.' + str(i)
                    fseg = f + i
                    if os.path.exists(os.path.join(src, fseg)):
                        if os.path.exists(os.path.join(dst,fseg)):
                            logger.debug('rm(%s)' % os.path.join(dst,fseg))
                            os.remove(os.path.join(dst,fseg))
                        logger.debug('cp(%s)' % os.path.join(dst,fseg))
                        shutil.copy2(os.path.join(src, fseg),
                                        os.path.join(dst, fseg))
                    else:
                        break
    #------------------------------------------------------------
    def DirIsIncludedFromBackup(self, d):
        if (d != 'base' and d != 'gpmigrator' and
            d != 'mirror_upgrade_state' and
            d.startswith('gp_dump_') == False and
            d.startswith('gp_cdatabase_') == False and
            d.startswith('pg_log') == False ):
            return True
        return False
    
    #------------------------------------------------------------
    def CreateMasterCopy(self):
        self.ForceShutDown()
        
        """
        Copy everthing under the master data, except base and gpmigrator, to gpmigrator/backup
        Also, exclude gp_dump_*, gp_cdatabase_* and pg_log
        """
        dst = os.path.join(self.masterdir, WORKDIR, BACKUPDIR)
        self.RunCmd('rm -rf ' + dst)
        os.mkdir(dst)
        for d in os.listdir(self.masterdir):
            if (self.DirIsIncludedFromBackup(d)):
                self.do_copy(self.masterdir, dst, d)

        """
        Copy everything under catdir, except the backup dir.
        """
        for src in self.mastercatdir:
            dst = os.path.join(src, BACKUPDIR)
            self.RunCmd('rm -rf ' + dst)
            os.mkdir(dst)
            for d in os.listdir(src):
                if (d != BACKUPDIR):
                    self.do_copy(src, dst, d)

    #------------------------------------------------------------
    def DeleteMasterCatalogCopy(self):
        # just print all the cat dirs
        logger.info("Deleting Master Catalog Backup Copy: %s" % self.mastercatdir)
        
        """
        Delete catdir/backup
        """
        for src in self.mastercatdir:
            dst = os.path.join(src, BACKUPDIR)
            self.RunCmd('rm -rf ' + dst)

        return "DeleteMasterDataDirCopy"

    #------------------------------------------------------------
    def RevertMasterCopyAndCatVersion(self):
        self.ForceShutDown()
        
        """
        Delete everything which doesn't exists in the backup dir
        Copy the backup to the cat dir
        """
        for catdir in self.mastercatdir:
            backupdir = os.path.join(catdir, BACKUPDIR)
            for d in os.listdir(catdir):
                if (d != BACKUPDIR):
                    backuploc = os.path.join(backupdir, d)
                    if (os.path.exists(backuploc)):
                        self.do_copy(backupdir, catdir, d)
                    else:
                        dst = os.path.join(catdir, d)
                        self.RunCmd('rm -rf ' + dst)

        """
        Delete everything under the master data, except base and gpmigrator, 
        mirror_upgrade_state file, and pg_log dir
        to gpmigrator/backup
        Copy things from backup
        """
        backupdir = os.path.join(self.masterdir, WORKDIR, BACKUPDIR)
        for d in os.listdir(self.masterdir):
            if (self.DirIsIncludedFromBackup(d)):
                backuploc = os.path.join(self.masterdir, WORKDIR, BACKUPDIR, d)
                if (os.path.exists(backuploc)):
                    self.do_copy(backupdir, self.masterdir, d)
                else:
                    dst = os.path.join(self.masterdir, d)
                    self.RunCmd('rm -rf ' + dst)

        """
        Set the catversion to the old one
        """
        self.SetCatVersion(False)
        
        return "SwitchToMirror"

    #------------------------------------------------------------
    def SwitchToMirror(self):
        # Set the mirror as primary
        self.ForceShutDown()
        self.Startup(self.oldenv, True)
        self.Update("""begin;
            set allow_system_table_mods='dml';
            update gp_segment_configuration set mode='c', role='p' where preferred_role='m';
            update gp_segment_configuration set mode='s', role='m', status='d' where preferred_role='p' and content >=0;
            commit;
        """)
        self.Shutdown()
        
        return "RecoverPrimary"

    #------------------------------------------------------------
    def RecoverPrimary(self):
        self.ForceShutDown()
        self.Startup(self.oldenv)
        logger.info("Running gprecoverseg to recover primary....")
        
        cmd = "$GPHOME/bin/gprecoverseg -a -F -l %s" % self.logdir
        p = subprocess.Popen(cmd, shell = True,
                            close_fds = True,
                            stdout = subprocess.PIPE,
                            stderr = subprocess.PIPE,
                            env=self.oldenv)
        result = p.communicate()
        if p.returncode != 0:
            logger.debug(result[0])
            logger.warn("****************************************************")
            logger.warn("Recovering Primary Failed...!!!!")
            logger.warn("   Please examine the log in %s" % self.logdir)
            logger.warn("   fix the problem and gprecoverseg -F to recover the primary")
            logger.warn("****************************************************")
        
        return "AbortMirrorUpgrade"


    #------------------------------------------------------------
    def DeleteMasterDataDirCopy(self):
        # just print all the cat dirs and data dir
        logger.info("Deleting Master Data Dir Backup Copy: %s" % self.masterdir)
        
        """
        Delete the work dir
        """
        dst = os.path.join(self.masterdir, WORKDIR)
        self.RunCmd('rm -rf ' + dst)

        return "FinishMirrorUpgrade"
        
    #------------------------------------------------------------
    def TurnOffMirror(self):
        # Check that mirror and primary are in sync again..!!!
        # Set the mirror off and put the primary on changetracking
        self.ForceShutDown()
        self.Startup(self.oldenv, True)
        
        nonsynccnt = self.Select('''select count(*)
             from pg_catalog.gp_segment_configuration
             where status != 'u' or mode != 's' or role != preferred_role
            ''' )
        if (nonsynccnt[0] > 0):
            raise UpgradeError("Mirror or Segments are down or out of sync")

        self.Update("""begin;
            set allow_system_table_mods='dml';
            update gp_segment_configuration set mode='c' where preferred_role='p' and content >=0;
            update gp_segment_configuration set status='d' where preferred_role='m';
            commit;
        """)
        self.Shutdown()
        return "CatUpgrade"

    #------------------------------------------------------------
    def CatUpgrade(self):
        newversion = GpVersion(self.getversion(self.newhome, self.newenv))
        snv        = newversion.getVersionRelease()
        
        self.ForceShutDown()
        
        try:
            # Upgrade pg_control first, and start up the database with
            # the upgrade flag on.
            self.SetCatVersion(True)
            self.Startup(self.newenv, False, True)
            runfile = "%s/share/postgresql/upgrade/upg2_catupgrade.sql" % self.newhome
            self.ExecuteSQLFileOnAllDBs(runfile)
            self.Shutdown()
            self.Startup(self.newenv, False, False)
            self.PerformPostUpgrade()
        except:
            logger.error(traceback.format_exc())
            return "RevertMasterCopyAndCatVersion"
        
        return "RecoverMirror"

    #------------------------------------------------------------
    def RecoverMirror(self):
        self.ForceShutDown()
        self.Startup(self.newenv)
        logger.info("Running gprecoverseg to recover mirror....")

        cmd = ' '.join(["$GPHOME/bin/gprecoverseg", '-a', '-l', self.logdir])
        p = subprocess.Popen(cmd, shell = True,
                            close_fds = True,
                            stdout = subprocess.PIPE,
                            stderr = subprocess.PIPE,
                            env=self.newenv)
        result = p.communicate()
        
        if p.returncode != 0:
            logger.debug(result[0])
            logger.warn("****************************************************")
            logger.warn("Recovering Mirror Failed...!!!!")
            logger.warn("   Please examine the log in %s" % self.logdir)
            logger.warn("   fix the problem and re-run gpmigrator_mirror to recover the mirror")
            logger.warn("****************************************************")

        return "DeleteMasterCatalogCopy"
    
    #------------------------------------------------------------
    def LoadState(self):
        # Setup the current state and statefile
        #   if state file exists, the current state is the last state
        self.mirror_upgrade_state = "CreateMirrorWorkDir"
        statefile                 = os.path.join(self.masterdir, 'mirror_upgrade_state')
        hasconfigfile             = False
        if os.path.exists(statefile):
            logger.info("Found previous upgrade")
            self.mirror_upgrade_state_file = open(statefile, 'r')
            for line in self.mirror_upgrade_state_file:
                line = line.strip()
                if (line == "TurnOffMirror"):
                    hasconfigfile = True
                if (line == 'DeleteMasterDataDirCopy'):
                    hasconfigfile = False
                self.mirror_upgrade_state = line
            self.mirror_upgrade_state_file.close()
        self.mirror_upgrade_state_file = open(statefile, 'a')
        
        if (hasconfigfile):
            self.ReadInfo()

    #------------------------------------------------------------
    def AbortMirrorUpgrade(self):
        self.mirror_upgrade_state_file.close()
        statefile = os.path.join(self.masterdir, 'mirror_upgrade_state')
        open(statefile, 'w').truncate()

        logger.info("---------------------------------------------------")
        logger.info("Upgrade Aborted")
        logger.info("---------------------------------------------------")
        logger.info("Please fix up the error and re-run upgrade")
        logger.info("---------------------------------------------------")

        sys.exit(0)
    #------------------------------------------------------------
    def FinishMirrorUpgrade(self):
        self.mirror_upgrade_state_file.close()
        statefile = os.path.join(self.masterdir, 'mirror_upgrade_state')
        self.RunCmd('rm -rf ' + statefile)
        
        logger.info("---------------------------------------------------")
        logger.info("Upgrade Successful")
        logger.info("---------------------------------------------------")
        logger.info("Please consult release notes for post-upgrade ")
        logger.info("instructions to complete environment configuration")
        logger.info("---------------------------------------------------")

        sys.exit(0)

    #------------------------------------------------------------
    def PerformMirroredUpgrade(self):
        #==========================================================
        # Upgrade strategy
        #   Use mirror as backup: start the db with mirror off and put primary in changetracking.
        #   Then start the db in upgrade mode to run the upgrade script as usual.
        #   When the script finished, use gprecoverseg to bring the mirror back in sync.
        #   In case of error when running the upgrade script, set the primary as down and bring
        #   the mirror up as primary. Do either full recovery (gprecoverseg -F) or delta
        #   (gpcheckmirrorseg, gprepairmirrorseg and gprecoverseg).
        #
        #   Master is not mirrored. We'll copyed everything in the master to a backup location
        #==========================================================

        # We use a state machine approach to handle fault and crash.
        # Fault will be cought as exception and it'll go the fault hanlding
        # state.
        # Crash will not change the state. When restart, it'll re-do the current state
        # agagin.

        # Initialize the current state 
        #   if a state file already exists, the current state is in the state file
        self.LoadState()
        
        faultinject_index = 100
        
        while True:
            logger.info("  ")
            logger.info("**************************************************")
            logger.info("**************************************************")
            logger.info("Executing state:%s"%self.mirror_upgrade_state)
            logger.info("**************************************************")
            logger.info("**************************************************")
            if (self.mirror_upgrade_state == "CreateMirrorWorkDir"):
                self.mirror_upgrade_state = self.CreateMirrorWorkDir()
            elif (self.mirror_upgrade_state == "ExtractInfoAndCopyMaster"):
                self.mirror_upgrade_state = self.ExtractInfoAndCopyMaster()
            elif (self.mirror_upgrade_state == "TurnOffMirror"):
                self.mirror_upgrade_state = self.TurnOffMirror()
            elif (self.mirror_upgrade_state == "CatUpgrade"):
                self.mirror_upgrade_state = self.CatUpgrade()
            elif (self.mirror_upgrade_state == "RecoverMirror"):
                self.mirror_upgrade_state = self.RecoverMirror()
            elif (self.mirror_upgrade_state == "DeleteMasterCatalogCopy"):
                self.mirror_upgrade_state = self.DeleteMasterCatalogCopy()
            elif (self.mirror_upgrade_state == "DeleteMasterDataDirCopy"):
                self.mirror_upgrade_state = self.DeleteMasterDataDirCopy()
            elif (self.mirror_upgrade_state == "RevertMasterCopyAndCatVersion"):
                self.mirror_upgrade_state = self.RevertMasterCopyAndCatVersion()
            elif (self.mirror_upgrade_state == "SwitchToMirror"):
                self.mirror_upgrade_state = self.SwitchToMirror()
            elif (self.mirror_upgrade_state == "RecoverPrimary"):
                self.mirror_upgrade_state = self.RecoverPrimary()
            elif (self.mirror_upgrade_state == "AbortMirrorUpgrade"):
                self.mirror_upgrade_state = self.AbortMirrorUpgrade()
            elif (self.mirror_upgrade_state == "FinishMirrorUpgrade"):
                self.mirror_upgrade_state = self.FinishMirrorUpgrade()
            else:
                # can't get here
                raise UpgradeError('Unknown Command : %s' % self.mirror_upgrade_state)
                sys.exit(1)
                
            if self.faultinjection == faultinject_index:
                    raise UpgradeError("faultinjection=%d" % self.faultinjection)
            faultinject_index = faultinject_index+1

            self.mirror_upgrade_state_file.write("%s\n" % self.mirror_upgrade_state)
            self.mirror_upgrade_state_file.flush()
        sys.exit(0)


#============================================================
if __name__ == '__main__':
    coverage = GpCoverage()
    coverage.start()
    
    # Create a new GPUpgrade - should never throw exception
    u = GPUpgrade()
    
    # parses and validates input
    try:
        u.Setup()
    except Exception, e:
        logger.fatal(str(e))
        sys.exit(1)
    
    # Execute the main upgrade routine, the Run() function itself is
    # just a switch on the "cmd", which switches between the main
    # PerformUpgrade() function and all the various subcommands that
    # are upgraded by backend executions.
    try:
        u.Run()
    # Any error that can happen
    except KeyboardInterrupt:
        logger.fatal('***************************************')
        logger.fatal('==  Upgrade interrupted by user ==        <<<<< ')
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    except ConnectionError, e:
        logger.fatal('***************************************')
        logger.fatal(str(e))
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    except UpgradeError, e:
        logger.fatal('***************************************')
        logger.fatal(str(e))
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    except CmdError, e:
        logger.fatal('***************************************')
        logger.fatal('Error executing command:')
        logger.fatal('  ' + str(e.cmd.strip()))
        logger.fatal('  ' + str(e.stderr.strip()))
        logGenericErrorMessage()
        logger.fatal('***************************************')
        sys.exit(1)
    
    # Shouldn't get any of these, if they occur it is probably a bug in
    # the upgrader.
    except Exception, e:
        logger.fatal('***************************************')
        logger.fatal(str(e))
        logger.fatal('***************************************')
        logger.fatal(traceback.format_exc())
        sys.exit(1)
    finally:
        coverage.stop()
        coverage.generate_report()

    sys.exit(0)
