diff --git a/gpMgmt/bin/gppylib/commands/pg.py b/gpMgmt/bin/gppylib/commands/pg.py index 9ef6acf8c611..fe81b673bad0 100644 --- a/gpMgmt/bin/gppylib/commands/pg.py +++ b/gpMgmt/bin/gppylib/commands/pg.py @@ -280,6 +280,33 @@ def create_slot(self): format(self.name, self.host, self.port)) return True +class PgReplicationSlotCopy: + def __init__(self, src_slot, dst_slot_name): + self.src_slot = src_slot + self.dst_slot_name = dst_slot_name + + def do_copy(self): + if not self.src_slot.slot_exists(): + logger.exception("PgReplicationSlotCopy: src slot {} doesn't exist for host:{}, port:{}".format(self.src_slot.name, self.src_slot.host, self.src_slot.port)) + return False + else: + logger.debug("Copy slot {} to {} for host:{}, port:{}".format(self.src_slot.name, self.dst_slot_name, self.src_slot.host, self.src_slot.port)) + sql = "SELECT pg_copy_physical_replication_slot('{}', '{}', false);".format(self.src_slot.name, self.dst_slot_name) + try: + dburl = dbconn.DbURL(hostname=self.src_slot.host, port=self.src_slot.port) + with closing(dbconn.connect(dburl, utility=True, encoding='UTF8')) as conn: + dbconn.query(conn, sql) + except DatabaseError as e: + logger.exception("Failed to query pg_copy_physical_replication_slot for host:{}, port:{}: {}". + format(self.src_slot.host, self.src_slot.port, str(e))) + return False + except Exception as ex: + raise Exception("Failed to create a copy of replication slot for host:{}, port:{} : {}". + format(self.src_slot.host, self.src_slot.port, str(ex))) + + logger.debug("Successfully created a copy {} of replication slot {} for host:{}, port:{}". + format(self.dst_slot_name, self.src_slot.name, self.src_slot.host, self.src_slot.port)) + return True class PgControlData(Command): def __init__(self, name, datadir, ctxt=LOCAL, remoteHost=None): diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 9c07e7f2387b..6cdbb5c67b8d 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -232,6 +232,17 @@ def _cleanup_before_recovery(self, gpArray, gpEnv): self._clean_up_failed_segments() self._set_seg_status_in_gparray() + def _cleanup_before_recovery_2(self, gpArray, gpEnv): + self.__logger.info("[RELOG] _stop_failed_segments - start") + self._stop_failed_segments(gpEnv) + self.__logger.info("[RELOG] _stop_failed_segments - end") + self.__logger.info("[RELOG] _wait_fts_to_mark_down_segments - start") + self._wait_fts_to_mark_down_segments(gpEnv, self._get_segments_to_mark_down()) + self.__logger.info("[RELOG] _wait_fts_to_mark_down_segments - end") + if not self.__forceoverwrite: + self._clean_up_failed_segments() + self._set_seg_status_in_gparray() + def _get_segments_to_mark_down(self): segments_to_mark_down = [] for toRecover in self.__mirrorsToBuild: @@ -275,32 +286,94 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): from the mirrorsToBuild must be present in gpArray. """ + self.__logger.info("[RELOG] __build_mirrors - start") if len(self.__mirrorsToBuild) == 0: self.__logger.info("No segments to {}".format(actionName)) return True + is_full_sync = True + for mirror in self.__mirrorsToBuild: + if not mirror.isFullSynchronization(): + is_full_sync = False + break + + is_dir_conflict = False + for mirror in self.__mirrorsToBuild: + if is_dir_conflict: + break; + failedSegment = mirror.getFailedSegment() + if failedSegment is None: + continue + for segmentPair in gpArray.getSegmentList(): + primaryDB = segmentPair.primaryDB + mirrorDB = segmentPair.mirrorDB + if primaryDB is not None: + if (primaryDB.hostname == failedSegment.hostname or primaryDB.address == failedSegment.address) and \ + primaryDB.datadir == failedSegment.datadir: + is_dir_conflict = True + break; + if mirrorDB is not None: + if (mirrorDB.hostname == failedSegment.hostname or mirrorDB.address == failedSegment.address) and \ + mirrorDB.datadir == failedSegment.datadir: + is_dir_conflict = True + break; + if actionName not in [GpMirrorListToBuild.Action.ADDMIRRORS, GpMirrorListToBuild.Action.RECOVERMIRRORS]: raise Exception('Invalid action. Valid values are {} and {}'.format(GpMirrorListToBuild.Action.RECOVERMIRRORS, GpMirrorListToBuild.Action.ADDMIRRORS)) self.__logger.info("%s segment(s) to %s" % (len(self.__mirrorsToBuild), actionName)) - self._cleanup_before_recovery(gpArray, gpEnv) - self._validate_gparray(gpArray) + recovery_result = False + + if is_full_sync and not is_dir_conflict: + self.checkForPortAndDirectoryConflicts(gpArray) - recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild) + self._validate_gparray(gpArray) - self._run_setup_recovery(actionName, recovery_info_by_host) + recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild) + + self._run_setup_recovery(actionName, recovery_info_by_host) + + # 1 - do pg_pasebackup with slot name = 'internal_wal_replication_slot_temp' + recovery_results_stage_1 = self._run_recovery_stage_1_basebackup(actionName, recovery_info_by_host, gpEnv) + + # 2 - Stop old mirrors + self._cleanup_before_recovery_2(gpArray, gpEnv) + + backout_map = self._update_config(recovery_info_by_host, gpArray) + + # 5 - Handle replication slots and start new mirrors + recovery_results_stage_2 = self._run_recovery_stage_2_start_segments(actionName, recovery_info_by_host, gpEnv) + + if actionName == GpMirrorListToBuild.Action.RECOVERMIRRORS: + self._revert_config_update(recovery_results_stage_1, backout_map) + + self._trigger_fts_probe(port=gpEnv.getCoordinatorPort()) + + self.__logger.info("[RELOG] __build_mirrors - end (segments are up)") + + recovery_result = recovery_results_stage_1.recovery_successful() and recovery_results_stage_2.recovery_successful() + + else: + self._cleanup_before_recovery(gpArray, gpEnv) + self._validate_gparray(gpArray) - backout_map = self._update_config(recovery_info_by_host, gpArray) + recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild) - recovery_results = self._run_recovery(actionName, recovery_info_by_host, gpEnv) - if actionName == GpMirrorListToBuild.Action.RECOVERMIRRORS: - self._revert_config_update(recovery_results, backout_map) + self._run_setup_recovery(actionName, recovery_info_by_host) - self._trigger_fts_probe(port=gpEnv.getCoordinatorPort()) + backout_map = self._update_config(recovery_info_by_host, gpArray) - return recovery_results.recovery_successful() + recovery_results = self._run_recovery(actionName, recovery_info_by_host, gpEnv) + if actionName == GpMirrorListToBuild.Action.RECOVERMIRRORS: + self._revert_config_update(recovery_results, backout_map) + + self._trigger_fts_probe(port=gpEnv.getCoordinatorPort()) + + recovery_result = recovery_results.recovery_successful() + + return recovery_result def _trigger_fts_probe(self, port=0): self.__logger.info('Triggering FTS probe') @@ -587,6 +660,23 @@ def _run_recovery(self, action_name, recovery_info_by_host, gpEnv): self._remove_progress_files(recovery_info_by_host, recovery_results) return recovery_results + def _run_recovery_stage_1_basebackup(self, action_name, recovery_info_by_host, gpEnv): + completed_recovery_results = self._do_recovery_stage_1_pg_basebackup(recovery_info_by_host, gpEnv) + recovery_results = RecoveryResult(action_name, completed_recovery_results, self.__logger) + recovery_results.print_bb_rewind_differential_update_and_start_errors() + + self._remove_progress_files(recovery_info_by_host, recovery_results) + return recovery_results + + def _run_recovery_stage_2_start_segments(self, action_name, recovery_info_by_host, gpEnv): + completed_recovery_results = self._do_recovery_stage_2_start_segments(recovery_info_by_host, gpEnv) + recovery_results = RecoveryResult(action_name, completed_recovery_results, self.__logger) + recovery_results.print_bb_rewind_differential_update_and_start_errors() + + self._remove_progress_files(recovery_info_by_host, recovery_results) + return recovery_results + + def _do_recovery(self, recovery_info_by_host, gpEnv): """ # Recover and start segments using gpsegrecovery, which will internally call either @@ -622,6 +712,65 @@ def _do_recovery(self, recovery_info_by_host, gpEnv): progressCmds=progress_cmds) return completed_recovery_results + def _do_recovery_stage_1_pg_basebackup(self, recovery_info_by_host, gpEnv): + """ + # do only pg_basebackup + """ + self.__logger.info('Initiating segment recovery - pg_basebackup') + cmds = [] + progress_cmds = [] + era = read_era(gpEnv.getCoordinatorDataDir(), logger=self.__logger) + for hostName, recovery_info_list in recovery_info_by_host.items(): + for ri in recovery_info_list: + ri.is_only_basebackup = True + ri.is_only_start = False + progressCmd = self._get_progress_cmd(ri.progress_file, ri.target_segment_dbid, hostName, ri.is_differential_recovery) + if progressCmd: + progress_cmds.append(progressCmd) + + cmds.append(gp.GpSegRecovery('Recover segments - Stage 1 - pg_basebackup', + recoveryinfo.serialize_list(recovery_info_list), + gplog.get_logger_dir(), + verbose=gplog.logging_is_verbose(), + batchSize=self.__parallelPerHost, + remoteHost=hostName, + era=era, + maxRate=self.__maxRate, + forceoverwrite=self.__forceoverwrite)) + completed_recovery_results = self.__runWaitAndCheckWorkerPoolForErrorsAndClear(cmds, suppressErrorCheck=True, + progressCmds=progress_cmds) + return completed_recovery_results + + def _do_recovery_stage_2_start_segments(self, recovery_info_by_host, gpEnv): + """ + # do only start of segments + """ + self.__logger.info('Initiating segment recovery - start segments') + cmds = [] + progress_cmds = [] + era = read_era(gpEnv.getCoordinatorDataDir(), logger=self.__logger) + for hostName, recovery_info_list in recovery_info_by_host.items(): + for ri in recovery_info_list: + ri.is_only_basebackup = False + ri.is_only_start = True + progressCmd = self._get_progress_cmd(ri.progress_file, ri.target_segment_dbid, hostName, ri.is_differential_recovery) + if progressCmd: + progress_cmds.append(progressCmd) + + cmds.append(gp.GpSegRecovery('Recover segments - Stage 2 - start segments', + recoveryinfo.serialize_list(recovery_info_list), + gplog.get_logger_dir(), + verbose=gplog.logging_is_verbose(), + batchSize=self.__parallelPerHost, + remoteHost=hostName, + era=era, + maxRate=self.__maxRate, + forceoverwrite=self.__forceoverwrite)) + completed_recovery_results = self.__runWaitAndCheckWorkerPoolForErrorsAndClear(cmds, suppressErrorCheck=True, + progressCmds=progress_cmds) + return completed_recovery_results + + def _do_setup_for_recovery(self, recovery_info_by_host): self.__logger.info('Setting up the required segments for recovery') cmds = [] diff --git a/gpMgmt/bin/gppylib/recoveryinfo.py b/gpMgmt/bin/gppylib/recoveryinfo.py index 9b03287a23ff..670dfc2ee3a3 100644 --- a/gpMgmt/bin/gppylib/recoveryinfo.py +++ b/gpMgmt/bin/gppylib/recoveryinfo.py @@ -15,7 +15,7 @@ class RecoveryInfo(object): Note: we don't have target hostname, since an object of this class will be accessed by the target host directly """ def __init__(self, target_datadir, target_port, target_segment_dbid, source_hostname, source_port, - source_datadir, is_full_recovery, is_differential_recovery, progress_file): + source_datadir, is_full_recovery, is_differential_recovery, progress_file, is_only_basebackup, is_only_start): self.target_datadir = target_datadir self.target_port = target_port self.target_segment_dbid = target_segment_dbid @@ -29,6 +29,8 @@ def __init__(self, target_datadir, target_port, target_segment_dbid, source_host self.is_full_recovery = is_full_recovery self.is_differential_recovery = is_differential_recovery self.progress_file = progress_file + self.is_only_basebackup = is_only_basebackup + self.is_only_start = is_only_start def __str__(self): return json.dumps(self, default=lambda o: o.__dict__) @@ -71,7 +73,7 @@ def build_recovery_info(mirrors_to_build): target_segment.getSegmentDataDirectory(), target_segment.getSegmentPort(), target_segment.getSegmentDbId(), source_segment.getSegmentHostName(), source_segment.getSegmentPort(), source_segment.getSegmentDataDirectory(), - to_recover.isFullSynchronization(), to_recover.isDifferentialSynchronization(), progress_file)) + to_recover.isFullSynchronization(), to_recover.isDifferentialSynchronization(), progress_file, False, False)) return recovery_info_by_host diff --git a/gpMgmt/sbin/gpsegrecovery.py b/gpMgmt/sbin/gpsegrecovery.py index 74c70ef8e92b..fe458249dc48 100644 --- a/gpMgmt/sbin/gpsegrecovery.py +++ b/gpMgmt/sbin/gpsegrecovery.py @@ -5,7 +5,7 @@ from contextlib import closing from gppylib.recoveryinfo import RecoveryErrorType -from gppylib.commands.pg import PgBaseBackup, PgRewind, PgReplicationSlot +from gppylib.commands.pg import PgBaseBackup, PgRewind, PgReplicationSlot, PgReplicationSlotCopy from gppylib.commands.unix import Rsync from recovery_base import RecoveryBase, set_recovery_cmd_results from gppylib.commands.base import Command, LOCAL @@ -61,6 +61,72 @@ def run(self): self.error_type = RecoveryErrorType.START_ERROR start_segment(self.recovery_info, self.logger, self.era) +class FullRecovery2Phase(Command): + def __init__(self, name, recovery_info, forceoverwrite, logger, era, maxRate, is_only_basebackup, is_only_start): + self.name = name + self.recovery_info = recovery_info + self.replicationSlotName = 'internal_wal_replication_slot' + self.replicationSlotNameTemp = 'internal_wal_replication_slot_temp' + self.forceoverwrite = forceoverwrite + self.era = era + self.maxRate = maxRate + # FIXME test for this cmdstr. also what should this cmdstr be ? + cmdStr = '' + #cmdstr = 'TODO? : {} {}'.format(str(recovery_info), self.verbose) + Command.__init__(self, self.name, cmdStr) + #FIXME this logger has to come after the init and is duplicated in all the 4 classes + self.logger = logger + self.error_type = RecoveryErrorType.DEFAULT_ERROR + self.is_only_basebackup = is_only_basebackup + self.is_only_start = is_only_start + + @set_recovery_cmd_results + def run(self): + self.logger.info("[RELOG] Running FullRecovery2Phase") + if self.is_only_basebackup: + self.logger.info("[RELOG] Only pg_basebackup") + self.error_type = RecoveryErrorType.BASEBACKUP_ERROR + cmd = PgBaseBackup(self.recovery_info.target_datadir, + self.recovery_info.source_hostname, + str(self.recovery_info.source_port), + create_slot=True, + replication_slot_name=self.replicationSlotNameTemp, + forceoverwrite=self.forceoverwrite, + target_gp_dbid=self.recovery_info.target_segment_dbid, + progress_file=self.recovery_info.progress_file, + max_rate=self.maxRate) + self.logger.info("Running pg_basebackup with progress output temporarily in %s" % self.recovery_info.progress_file) + self.logger.info("[RELOG] pg_basebackup - start") + cmd.run(validateAfter=True) + self.logger.info("[RELOG] pg_basebackup - end") + self.error_type = RecoveryErrorType.DEFAULT_ERROR + self.logger.info("Successfully ran pg_basebackup for dbid: {}".format( + self.recovery_info.target_segment_dbid)) + if self.is_only_start: + # 3. drop replication slot 'internal_wal_replication_slot' + self.logger.info("[RELOG] drop old slot") + oldSlot = PgReplicationSlot(self.recovery_info.source_hostname, str(self.recovery_info.source_port), self.replicationSlotName) + if oldSlot.slot_exists(): + oldSlot.drop_slot() + + # 4. select pg_copy_physical_replication_slot('internal_wal_replication_slot_temp', 'internal_wal_replication_slot', false); + self.logger.info("[RELOG] Create a new replication slot") + tempSlot = PgReplicationSlot(self.recovery_info.source_hostname, str(self.recovery_info.source_port), self.replicationSlotNameTemp) + newSlot = PgReplicationSlotCopy(tempSlot, self.replicationSlotName) + newSlot.do_copy() + + self.logger.info("[RELOG] Only start of segments") + # Updating port number on conf after recovery + self.error_type = RecoveryErrorType.UPDATE_ERROR + update_port_in_conf(self.recovery_info, self.logger) + update_replication_slot_in_conf(self.recovery_info, self.logger, self.replicationSlotName) + self.error_type = RecoveryErrorType.START_ERROR + start_segment(self.recovery_info, self.logger, self.era) + + # 6. Drop temp replication slot + self.logger.info("[RELOG] drop temp replication slot") + tempSlot.drop_slot() + class IncrementalRecovery(Command): def __init__(self, name, recovery_info, logger, era): @@ -364,6 +430,14 @@ def update_port_in_conf(recovery_info, logger): 'port', recovery_info.target_port, optType='number') modifyConfCmd.run(validateAfter=True) +def update_replication_slot_in_conf(recovery_info, logger, slot): + logger.info("Updating %s/postgresql.auto.conf" % recovery_info.target_datadir) + modifyConfCmd = ModifyConfSetting('Updating %s/postgresql.auto.conf' % recovery_info.target_datadir, + "{}/{}".format(recovery_info.target_datadir, 'postgresql.auto.conf'), + 'primary_slot_name', slot, optType='string') + modifyConfCmd.run(validateAfter=True) + + #FIXME we may not need this class class SegRecovery(object): @@ -396,12 +470,22 @@ def get_recovery_cmds(self, seg_recovery_info_list, forceoverwrite, logger, era, cmd_list = [] for seg_recovery_info in seg_recovery_info_list: if seg_recovery_info.is_full_recovery: - cmd = FullRecovery(name='Run pg_basebackup', - recovery_info=seg_recovery_info, - forceoverwrite=forceoverwrite, - logger=logger, - era=era, - maxRate=maxRate) + if seg_recovery_info.is_only_basebackup or seg_recovery_info.is_only_start: + cmd = FullRecovery2Phase(name='Run 2-phase pg_basebackup', + recovery_info=seg_recovery_info, + forceoverwrite=forceoverwrite, + logger=logger, + era=era, + maxRate=maxRate, + is_only_basebackup=seg_recovery_info.is_only_basebackup, + is_only_start=seg_recovery_info.is_only_start) + else: + cmd = FullRecovery(name='Run pg_basebackup', + recovery_info=seg_recovery_info, + forceoverwrite=forceoverwrite, + logger=logger, + era=era, + maxRate=maxRate) elif seg_recovery_info.is_differential_recovery: cmd = DifferentialRecovery(name='Run rsync', recovery_info=seg_recovery_info, diff --git a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature index 9817183b5792..c7e0d5467188 100644 --- a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature +++ b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature @@ -21,8 +21,8 @@ Feature: Tests for gpaddmirrors When gpaddmirrors adds 3 mirrors with additional args "" Then gpaddmirrors should only spawn up to workers in WorkerPool And check if gpaddmirrors ran "$GPHOME/sbin/gpsegsetuprecovery.py" 1 times with args "-b " - And check if gpaddmirrors ran "$GPHOME/sbin/gpsegrecovery.py" 1 times with args "-b " - And check if gpaddmirrors ran "$GPHOME/sbin/gpsegrecovery.py" 1 times with args "-b " + And check if gpaddmirrors ran "$GPHOME/sbin/gpsegrecovery.py" 2 times with args "-b " + And check if gpaddmirrors ran "$GPHOME/sbin/gpsegrecovery.py" 2 times with args "-b " And an FTS probe is triggered And the segments are synchronized And verify the database has mirrors diff --git a/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature b/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature index eadafcdf9928..b2a4dc8be0f1 100644 --- a/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature +++ b/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature @@ -203,7 +203,6 @@ Feature: Tests for gpmovemirrors And edit the input file to recover mirror with content 0 to a new directory on remote host with mode 0700 And edit the input file to recover mirror with content 1 to a new directory on remote host with mode 0700 When the user asynchronously runs gpmovemirrors with input file and additional args " " and the process is saved - And the user waits until mirror on content 0,1 is down And the user suspend the walsender on the primary on content 0 Then the user waits until recovery_progress.file is created in gpAdminLogs and verifies its format And verify that lines from recovery_progress.file are present in segment progress files in gpAdminLogs @@ -227,7 +226,6 @@ Feature: Tests for gpmovemirrors And edit the input file to recover mirror with content 1 to a new directory on remote host with mode 0700 And edit the input file to recover mirror with content 2 to a new directory on remote host with mode 0700 When the user asynchronously runs gpmovemirrors with input file and additional args " " and the process is saved - And the user waits until mirror on content 0,1,2 is down And the user suspend the walsender on the primary on content 0 Then the user waits until recovery_progress.file is created in gpAdminLogs and verifies its format And verify that lines from recovery_progress.file are present in segment progress files in gpAdminLogs diff --git a/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature b/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature index 4afc5a9aa654..f0142f9bc135 100644 --- a/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature +++ b/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature @@ -208,7 +208,7 @@ Feature: gprecoverseg tests Then gprecoverseg should return a return code of 0 And gprecoverseg should only spawn up to workers in WorkerPool And check if gprecoverseg ran "$GPHOME/sbin/gpsegsetuprecovery.py" 1 times with args "-b " - And check if gprecoverseg ran "$GPHOME/sbin/gpsegrecovery.py" 1 times with args "-b " + And check if gprecoverseg ran "$GPHOME/sbin/gpsegrecovery.py" 2 times with args "-b " And gpsegsetuprecovery should only spawn up to workers in WorkerPool And gpsegrecovery should only spawn up to workers in WorkerPool And check if gprecoverseg ran "$GPHOME/sbin/gpsegstop.py" 1 times with args "-b "