| #!/usr/bin/env python3 |
| # Line too long - pylint: disable=C0301 |
| # Invalid name - pylint: disable=C0103 |
| # |
| # Copyright (c) EMC/Greenplum Inc 2011. All Rights Reserved. |
| # Copyright (c) Greenplum Inc 2008. All Rights Reserved. |
| # |
| """ |
| Internal Use Function. |
| """ |
| |
| # THIS IMPORT MUST COME FIRST |
| from gppylib.mainUtils import simple_main, addStandardLoggingAndHelpOptions |
| |
| import os, pickle, base64, time |
| |
| from gppylib.gpparseopts import OptParser, OptChecker |
| from gppylib import gparray, gplog |
| from gppylib.commands import base, gp |
| from gppylib.utils import parseKeyColonValueLines |
| from gppylib.commands.pg import PgControlData |
| |
| logger = gplog.get_default_logger() |
| |
| |
| DESCRIPTION = """ |
| This utility is NOT SUPPORTED and is for internal-use only. |
| Starts a set of one or more segment databases. |
| """ |
| |
| HELP = [""" |
| Utility should only be used by other GP utilities. |
| |
| Return codes: |
| 0 - All segments started successfully |
| 1 - At least one segment didn't start successfully |
| |
| """] |
| |
| |
| |
| |
| class StartResult: |
| """ |
| Recorded result information from an attempt to start one segment. |
| """ |
| |
| def __init__(self, datadir, started, reason, reasoncode): |
| """ |
| @param datadir |
| @param started |
| @param reason |
| @param reasoncode one of the gp.SEGSTART_* values |
| """ |
| self.datadir = datadir |
| self.started = started |
| self.reason = reason |
| self.reasoncode = reasoncode |
| |
| def __str__(self): |
| return "".join([ |
| "STATUS", |
| "--DIR:", str(self.datadir), |
| "--STARTED:", str(self.started), |
| "--REASONCODE:", str(self.reasoncode), |
| "--REASON:", str(self.reason) |
| ]) |
| |
| |
| class OverallStatus: |
| """ |
| Mapping and segment status information for all segments on this host. |
| """ |
| |
| def __init__(self, dblist): |
| """ |
| Build the datadir->segment mapping and remember the original size. |
| Since segments which fail to start will be removed from the mapping, |
| we later test the size of the map against the original size when |
| returning the appropriate status code to the caller. |
| """ |
| self.dirmap = dict([(seg.getSegmentDataDirectory(), seg) for seg in dblist]) |
| self.original_length = len(self.dirmap) |
| self.results = [] |
| self.logger = logger |
| |
| |
| def mark_failed(self, datadir, msg, reasoncode): |
| """ |
| Mark a segment as failed during some startup process. |
| Remove the entry for the segment from dirmap. |
| |
| @param datadir |
| @param msg |
| @param reasoncode one of the gp.SEGSTART_* constant values |
| """ |
| self.logger.info("Marking failed %s, %s, %s" % (datadir, msg, reasoncode)) |
| self.results.append( StartResult(datadir=datadir, started=False, reason=msg, reasoncode=reasoncode) ) |
| del self.dirmap[datadir] |
| |
| |
| def remaining_items_succeeded(self): |
| """ |
| Add results for all remaining items in our datadir->segment map. |
| """ |
| for datadir in list(self.dirmap.keys()): |
| self.results.append( StartResult(datadir=datadir, started=True, reason="Start Succeeded", reasoncode=gp.SEGSTART_SUCCESS ) ) |
| |
| |
| def log_results(self): |
| """ |
| Log info messages with our results |
| """ |
| status = '\nCOMMAND RESULTS\n' + "\n".join([str(result) for result in self.results]) |
| self.logger.info(status) |
| |
| |
| def exit_code(self): |
| """ |
| Return an appropriate exit code: 0 if no failures, 1 if some segments failed to start. |
| """ |
| if len(self.dirmap) != self.original_length: |
| return 1 |
| return 0 |
| |
| |
| |
| class GpSegStart: |
| """ |
| Logic to start segment servers on this host. |
| """ |
| |
| def __init__(self, dblist, gpversion, mirroringMode, num_cids, era, |
| timeout, pickledTransitionData, specialMode, wrapper, wrapper_args, |
| coordinator_checksum_version, segment_batch_size, logfileDirectory=False): |
| |
| # validate/store arguments |
| # |
| self.dblist = list(map(gparray.Segment.initFromString, dblist)) |
| |
| expected_gpversion = gpversion |
| actual_gpversion = gp.GpVersion.local('local software version check', os.path.abspath(os.pardir)) |
| if actual_gpversion != expected_gpversion: |
| raise Exception("Local Software Version does not match what is expected.\n" |
| "The local software version is: '%s'\n" |
| "But we were expecting it to be: '%s'\n" |
| "Please review and correct" % (actual_gpversion, expected_gpversion)) |
| |
| self.mirroringMode = mirroringMode |
| self.num_cids = num_cids |
| self.era = era |
| self.timeout = timeout |
| self.pickledTransitionData = pickledTransitionData |
| |
| assert(specialMode in [None, 'upgrade', 'maintenance']) |
| self.specialMode = specialMode |
| |
| self.wrapper = wrapper |
| self.wrapper_args = wrapper_args |
| |
| # initialize state |
| # |
| self.pool = base.WorkerPool(numWorkers=min(len(dblist), segment_batch_size)) |
| self.logger = logger |
| self.overall_status = None |
| |
| self.logfileDirectory = logfileDirectory |
| self.coordinator_checksum_version = coordinator_checksum_version |
| |
| def getOverallStatusKeys(self): |
| return list(self.overall_status.dirmap.keys()) |
| |
| # return True if all running |
| # return False if not all running |
| def checkPostmasters(self, must_be_running): |
| """ |
| Check that segment postmasters have been started. |
| @param must_be_running True if postmasters must be running by now. |
| """ |
| self.logger.info("Checking segment postmasters... (must_be_running %s)" % must_be_running) |
| |
| all_running = True |
| for datadir in self.getOverallStatusKeys(): |
| pid = gp.read_postmaster_pidfile(datadir) |
| running = gp.check_pid(pid) |
| msg = "Postmaster %s %srunning (pid %d)" % (datadir, "is " if running else "NOT ", pid) |
| self.logger.info(msg) |
| |
| if not running: |
| all_running = False |
| |
| if must_be_running and not running: |
| reasoncode = gp.SEGSTART_ERROR_PG_CTL_FAILED |
| self.overall_status.mark_failed(datadir, msg, reasoncode) |
| |
| return all_running |
| |
| |
| def __validateDirectoriesAndSetupRecoveryStartup(self): |
| """ |
| validate that the directories all exist and run recovery startup if needed |
| """ |
| self.logger.info("Validating directories...") |
| |
| for datadir in list(self.overall_status.dirmap.keys()): |
| self.logger.info("Validating directory: %s" % datadir) |
| |
| if os.path.isdir(datadir): |
| # |
| # segment datadir exists |
| # |
| postmaster_pid = os.path.join(datadir, 'postmaster.pid') |
| if os.path.exists(postmaster_pid): |
| self.logger.warning("postmaster.pid file exists, checking if recovery startup required") |
| |
| msg = gp.recovery_startup(datadir) |
| if msg: |
| reasoncode = gp.SEGSTART_ERROR_STOP_RUNNING_SEGMENT_FAILED |
| self.overall_status.mark_failed(datadir, msg, reasoncode) |
| |
| else: |
| # |
| # segment datadir does not exist |
| # |
| msg = "Segment data directory does not exist for: '%s'" % datadir |
| self.logger.warning(msg) |
| |
| reasoncode = gp.SEGSTART_ERROR_DATA_DIRECTORY_DOES_NOT_EXIST |
| self.overall_status.mark_failed(datadir, msg, reasoncode) |
| |
| |
| def __startSegments(self): |
| """ |
| Start the segments themselves |
| """ |
| self.logger.info("Starting segments... (mirroringMode %s)" % self.mirroringMode) |
| |
| for datadir, seg in list(self.overall_status.dirmap.items()): |
| |
| if self.coordinator_checksum_version != None: |
| cmd = PgControlData(name='run pg_controldata', datadir=datadir) |
| cmd.run(validateAfter=True) |
| res = cmd.get_results() |
| |
| if res.rc != 0: |
| msg = "pg_controldata failed.\nstdout:%s\nstderr:%s\n" % (res.stdout, res.stderr) |
| reasoncode = gp.SEGSTART_ERROR_PG_CONTROLDATA_FAILED |
| self.overall_status.mark_failed(datadir, msg, reasoncode) |
| continue |
| |
| segment_heap_checksum_version = cmd.get_value('Data page checksum version') |
| if segment_heap_checksum_version != self.coordinator_checksum_version: |
| msg = "Segment checksum %s does not match coordinator checksum %s.\n" % (segment_heap_checksum_version, |
| self.coordinator_checksum_version) |
| reasoncode = gp.SEGSTART_ERROR_CHECKSUM_MISMATCH |
| self.overall_status.mark_failed(datadir, msg, reasoncode) |
| continue |
| |
| cmd = gp.SegmentStart("Starting seg at dir %s" % datadir, |
| seg, |
| self.num_cids, |
| self.era, |
| self.mirroringMode, |
| timeout=self.timeout, |
| specialMode=self.specialMode, |
| wrapper=self.wrapper, |
| wrapper_args=self.wrapper_args) |
| self.pool.addCommand(cmd) |
| |
| self.pool.join() |
| |
| for cmd in self.pool.getCompletedItems(): |
| res = cmd.get_results() |
| if res.rc != 0: |
| |
| # we should also read in last entries in startup.log here |
| |
| datadir = cmd.segment.getSegmentDataDirectory() |
| msg = "PG_CTL failed.\nstdout:%s\nstderr:%s\n" % (res.stdout, res.stderr) |
| reasoncode = gp.SEGSTART_ERROR_PG_CTL_FAILED |
| self.overall_status.mark_failed(datadir, msg, reasoncode) |
| |
| self.pool.empty_completed_items() |
| |
| def run(self): |
| """ |
| Logic to start the segments. |
| """ |
| |
| # we initialize an overall status object which maintains a mapping |
| # from each segment's data directory to the segment object as well |
| # as a list of specific success/failure results. |
| # |
| self.overall_status = OverallStatus(self.dblist) |
| |
| # Each of the next four steps executes operations which may cause segment |
| # details to be removed from the mapping and recorded as failures. |
| # |
| self.__validateDirectoriesAndSetupRecoveryStartup() |
| self.__startSegments() |
| |
| # Being paranoid, we frequently check for postmaster failures. |
| # The postmasters should be running by now. |
| # |
| self.checkPostmasters(must_be_running=True) |
| |
| # At this point any segments remaining in the mapping are assumed to |
| # have successfully started. |
| # |
| self.overall_status.remaining_items_succeeded() |
| self.overall_status.log_results() |
| return self.overall_status.exit_code() |
| |
| |
| |
| def cleanup(self): |
| """ |
| Cleanup worker pool resources |
| """ |
| if self.pool: |
| self.pool.haltWork() |
| |
| |
| @staticmethod |
| def createParser(): |
| """ |
| Create parser expected by simple_main |
| """ |
| |
| parser = OptParser(option_class=OptChecker, |
| description=' '.join(DESCRIPTION.split()), |
| version='%prog version main build dev') |
| parser.setHelp(HELP) |
| |
| # |
| # Note that this mirroringmode parameter should only be either mirrorless or quiescent. |
| # If quiescent then it is implied that there is pickled transition data that will be |
| # provided (using -p) to immediately convert to a primary or a mirror. |
| # |
| addStandardLoggingAndHelpOptions(parser, includeNonInteractiveOption=False) |
| |
| parser.add_option("-D", "--dblist", dest="dblist", action="append", type="string") |
| parser.add_option("-M", "--mirroringmode", dest="mirroringMode", type="string") |
| parser.add_option("-p", "--pickledTransitionData", dest="pickledTransitionData", type="string") |
| parser.add_option("-V", "--gp-version", dest="gpversion", metavar="GP_VERSION", help="expected software version") |
| parser.add_option("-n", "--numsegments", dest="num_cids", help="number of distinct content ids in cluster") |
| parser.add_option("", "--era", dest="era", help="coordinator era") |
| parser.add_option("-t", "--timeout", dest="timeout", type="int", default=gp.SEGMENT_TIMEOUT_DEFAULT, |
| help="seconds to wait") |
| parser.add_option('-U', '--specialMode', type='choice', choices=['upgrade', 'maintenance'], |
| metavar='upgrade|maintenance', action='store', default=None, |
| help='start the instance in upgrade or maintenance mode') |
| parser.add_option('', '--wrapper', dest="wrapper", default=None, type='string') |
| parser.add_option('', '--wrapper-args', dest="wrapper_args", default=None, type='string') |
| parser.add_option('', '--coordinator-checksum-version', dest="coordinator_checksum_version", default=None, type='string', action="store") |
| parser.add_option('-b', '--segment-batch-size', type="int", dest="segment_batch_size", default=gp.DEFAULT_GPSTART_NUM_WORKERS, |
| help='Max number of segments per host to operate on in parallel.') |
| |
| return parser |
| |
| |
| @staticmethod |
| def createProgram(options, args): |
| """ |
| Create program expected by simple_main |
| """ |
| logfileDirectory = options.ensure_value("logfileDirectory", False) |
| return GpSegStart(options.dblist, |
| options.gpversion, |
| options.mirroringMode, |
| options.num_cids, |
| options.era, |
| options.timeout, |
| options.pickledTransitionData, |
| options.specialMode, |
| options.wrapper, |
| options.wrapper_args, |
| options.coordinator_checksum_version, |
| options.segment_batch_size, |
| logfileDirectory=logfileDirectory) |
| |
| #------------------------------------------------------------------------- |
| if __name__ == '__main__': |
| mainOptions = { 'setNonuserOnToolLogger':True} |
| simple_main( GpSegStart.createParser, GpSegStart.createProgram, mainOptions ) |
