From 9fdd5dd95a0601e70ce2e4aebdc0d1e6ca58008c Mon Sep 17 00:00:00 2001 From: Alexander Kondakov Date: Fri, 20 Dec 2024 05:53:13 +0300 Subject: [PATCH 01/10] Initial commit From f99f4ca2c6db68b0e713f3d040f496b9979905dd Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Wed, 22 Jan 2025 15:19:31 +1000 Subject: [PATCH 02/10] Decrease non-HA mode during gprecoverseg (quick & dirty proto) --- .../gppylib/operations/buildMirrorSegments.py | 101 +++++++++++++++++- gpMgmt/bin/gppylib/recoveryinfo.py | 6 +- gpMgmt/sbin/gpsegrecovery.py | 80 ++++++++++++-- 3 files changed, 174 insertions(+), 13 deletions(-) diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 9c07e7f2387b..0563070432c1 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -225,9 +225,13 @@ def getMaxTransferRate(self): return self.__maxRate def _cleanup_before_recovery(self, gpArray, gpEnv): - self.checkForPortAndDirectoryConflicts(gpArray) + # self.checkForPortAndDirectoryConflicts(gpArray) + 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() @@ -275,6 +279,7 @@ 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 @@ -285,22 +290,32 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): self.__logger.info("%s segment(s) to %s" % (len(self.__mirrorsToBuild), actionName)) - self._cleanup_before_recovery(gpArray, gpEnv) + #self._cleanup_before_recovery(gpArray, gpEnv) + self.checkForPortAndDirectoryConflicts(gpArray) + self._validate_gparray(gpArray) recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild) self._run_setup_recovery(actionName, recovery_info_by_host) + recovery_results_stage_1 = self._run_recovery_stage_1_basebackup(actionName, recovery_info_by_host, gpEnv) + + self._cleanup_before_recovery(gpArray, gpEnv) + backout_map = self._update_config(recovery_info_by_host, gpArray) - recovery_results = self._run_recovery(actionName, recovery_info_by_host, gpEnv) + 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, backout_map) + self._revert_config_update(recovery_results_stage_1, backout_map) self._trigger_fts_probe(port=gpEnv.getCoordinatorPort()) - return recovery_results.recovery_successful() + 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() + return recovery_result def _trigger_fts_probe(self, port=0): self.__logger.info('Triggering FTS probe') @@ -587,6 +602,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 +654,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..98bbc849dd7a 100644 --- a/gpMgmt/sbin/gpsegrecovery.py +++ b/gpMgmt/sbin/gpsegrecovery.py @@ -61,6 +61,56 @@ 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 + # TODO: deal with '_upd' suffix... + self.replicationSlotName = 'internal_wal_replication_slot_upd' + 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.replicationSlotName, + 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: + 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) + class IncrementalRecovery(Command): def __init__(self, name, recovery_info, logger, era): @@ -364,6 +414,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.conf" % recovery_info.target_datadir) + modifyConfCmd = ModifyConfSetting('Updating %s/postgresql.conf' % recovery_info.target_datadir, + "{}/{}".format(recovery_info.target_datadir, 'postgresql.conf'), + 'primary_slot_name', slot, optType='string') + modifyConfCmd.run(validateAfter=True) + + #FIXME we may not need this class class SegRecovery(object): @@ -396,12 +454,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, From e9e1a92b32e4bd2c0c1127a3e0080c3f3d41b1b5 Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Fri, 31 Jan 2025 23:23:26 +1000 Subject: [PATCH 03/10] Handle replication slot --- gpMgmt/bin/gppylib/commands/pg.py | 27 +++++++++++++++++ .../gppylib/operations/buildMirrorSegments.py | 3 ++ gpMgmt/sbin/gpsegrecovery.py | 30 ++++++++++++++----- 3 files changed, 53 insertions(+), 7 deletions(-) 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 0563070432c1..4ac55ed18af8 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -299,12 +299,15 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): 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(gpArray, gpEnv) backout_map = self._update_config(recovery_info_by_host, gpArray) + # 5 - 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: diff --git a/gpMgmt/sbin/gpsegrecovery.py b/gpMgmt/sbin/gpsegrecovery.py index 98bbc849dd7a..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 @@ -65,8 +65,8 @@ 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 - # TODO: deal with '_upd' suffix... - self.replicationSlotName = 'internal_wal_replication_slot_upd' + self.replicationSlotName = 'internal_wal_replication_slot' + self.replicationSlotNameTemp = 'internal_wal_replication_slot_temp' self.forceoverwrite = forceoverwrite self.era = era self.maxRate = maxRate @@ -90,7 +90,7 @@ def run(self): self.recovery_info.source_hostname, str(self.recovery_info.source_port), create_slot=True, - replication_slot_name=self.replicationSlotName, + replication_slot_name=self.replicationSlotNameTemp, forceoverwrite=self.forceoverwrite, target_gp_dbid=self.recovery_info.target_segment_dbid, progress_file=self.recovery_info.progress_file, @@ -103,6 +103,18 @@ def run(self): 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 @@ -111,6 +123,10 @@ def run(self): 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): @@ -415,9 +431,9 @@ def update_port_in_conf(recovery_info, logger): modifyConfCmd.run(validateAfter=True) def update_replication_slot_in_conf(recovery_info, logger, slot): - logger.info("Updating %s/postgresql.conf" % recovery_info.target_datadir) - modifyConfCmd = ModifyConfSetting('Updating %s/postgresql.conf' % recovery_info.target_datadir, - "{}/{}".format(recovery_info.target_datadir, 'postgresql.conf'), + 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) From a0421991ce8423c66004f921ab18bc2018fb3672 Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Mon, 3 Feb 2025 10:40:25 +1000 Subject: [PATCH 04/10] Update to fix non-full sync --- .../gppylib/operations/buildMirrorSegments.py | 70 ++++++++++++++----- 1 file changed, 52 insertions(+), 18 deletions(-) diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 4ac55ed18af8..5599c58ae6a8 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -225,7 +225,14 @@ def getMaxTransferRate(self): return self.__maxRate def _cleanup_before_recovery(self, gpArray, gpEnv): - # self.checkForPortAndDirectoryConflicts(gpArray) + self.checkForPortAndDirectoryConflicts(gpArray) + self._stop_failed_segments(gpEnv) + self._wait_fts_to_mark_down_segments(gpEnv, self._get_segments_to_mark_down()) + if not self.__forceoverwrite: + 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") @@ -284,40 +291,67 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): self.__logger.info("No segments to {}".format(actionName)) return True + is_full_sync = False + for mirror in self.__mirrorsToBuild: + if mirror.isFullSynchronization(): + is_full_sync = 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.checkForPortAndDirectoryConflicts(gpArray) + recovery_result = False - self._validate_gparray(gpArray) + if is_full_sync: + 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) - # 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) + 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) - # 2 - Stop old mirrors - self._cleanup_before_recovery(gpArray, gpEnv) + recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild) - backout_map = self._update_config(recovery_info_by_host, gpArray) + self._run_setup_recovery(actionName, recovery_info_by_host) - # 5 - Start new mirrors - recovery_results_stage_2 = self._run_recovery_stage_2_start_segments(actionName, recovery_info_by_host, gpEnv) + backout_map = self._update_config(recovery_info_by_host, gpArray) - if actionName == GpMirrorListToBuild.Action.RECOVERMIRRORS: - self._revert_config_update(recovery_results_stage_1, backout_map) + 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()) + self._trigger_fts_probe(port=gpEnv.getCoordinatorPort()) - self.__logger.info("[RELOG] __build_mirrors - end (segments are up)") + recovery_result = recovery_results.recovery_successful() - recovery_result = recovery_results_stage_1.recovery_successful() and recovery_results_stage_2.recovery_successful() return recovery_result def _trigger_fts_probe(self, port=0): From cfe99fcd40ac01fae2784b631892b67ab9c41b67 Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Mon, 3 Feb 2025 15:35:07 +1000 Subject: [PATCH 05/10] Additional fixes for prototype --- .../bin/gppylib/operations/buildMirrorSegments.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 5599c58ae6a8..b7a083a64718 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -297,6 +297,18 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): is_full_sync = True break + is_dir_conflict = False + for mirror in self.__mirrorsToBuild: + failedSegment = mirror.getFailedSegment() + for segmentPair in gpArray.getSegmentList(): + if ((segmentPair.primaryDB.hostname == failedSegment.hostname or \ + segmentPair.primaryDB.address == failedSegment.address) and \ + segmentPair.primaryDB.datadir == failedSegment.datadir) or \ + ((segmentPair.mirrorDB.hostname == failedSegment.hostname or \ + segmentPair.mirrorDB.address == failedSegment.address) and \ + segmentPair.mirrorDB.datadir == failedSegment.datadir): + is_dir_conflict = True + 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)) @@ -305,7 +317,7 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): recovery_result = False - if is_full_sync: + if is_full_sync and not is_dir_conflict: self.checkForPortAndDirectoryConflicts(gpArray) self._validate_gparray(gpArray) From 295a13e28661f4ce989d25d29934a3a41873d37d Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Mon, 3 Feb 2025 22:34:04 +1000 Subject: [PATCH 06/10] Updates --- .../gppylib/operations/buildMirrorSegments.py | 27 ++++++++++++------- 1 file changed, 17 insertions(+), 10 deletions(-) diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index b7a083a64718..4b670d181eab 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -291,23 +291,30 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): self.__logger.info("No segments to {}".format(actionName)) return True - is_full_sync = False + is_full_sync = True for mirror in self.__mirrorsToBuild: - if mirror.isFullSynchronization(): - is_full_sync = True + 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() for segmentPair in gpArray.getSegmentList(): - if ((segmentPair.primaryDB.hostname == failedSegment.hostname or \ - segmentPair.primaryDB.address == failedSegment.address) and \ - segmentPair.primaryDB.datadir == failedSegment.datadir) or \ - ((segmentPair.mirrorDB.hostname == failedSegment.hostname or \ - segmentPair.mirrorDB.address == failedSegment.address) and \ - segmentPair.mirrorDB.datadir == failedSegment.datadir): - is_dir_conflict = True + 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, From ac0c6201a02f536e5b64534b7e835f10bd74962e Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Tue, 4 Feb 2025 08:33:16 +1000 Subject: [PATCH 07/10] Update --- gpMgmt/bin/gppylib/operations/buildMirrorSegments.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 4b670d181eab..6cdbb5c67b8d 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -302,6 +302,8 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): if is_dir_conflict: break; failedSegment = mirror.getFailedSegment() + if failedSegment is None: + continue for segmentPair in gpArray.getSegmentList(): primaryDB = segmentPair.primaryDB mirrorDB = segmentPair.mirrorDB From 7a2674969c5f9461c7cc0b12d47d3cc7646688af Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Tue, 4 Feb 2025 14:57:05 +1000 Subject: [PATCH 08/10] Attempt to fix tests --- gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature | 2 +- gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature index 9817183b5792..67e004cdeded 100644 --- a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature +++ b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature @@ -21,7 +21,7 @@ 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" 2 times with args "-b " And check if gpaddmirrors ran "$GPHOME/sbin/gpsegrecovery.py" 1 times with args "-b " And an FTS probe is triggered And the segments are synchronized 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 " From 16159dcbe423b70bd4c5632667e4cc29a99e5b62 Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Tue, 4 Feb 2025 18:39:31 +1000 Subject: [PATCH 09/10] Fix tests --- gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature index 67e004cdeded..c7e0d5467188 100644 --- a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature +++ b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature @@ -22,7 +22,7 @@ Feature: Tests for gpaddmirrors 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" 2 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 an FTS probe is triggered And the segments are synchronized And verify the database has mirrors From f0b5bb61a3a16a7422318d1a865cf6f4da62bbdd Mon Sep 17 00:00:00 2001 From: Roman Eskin Date: Wed, 5 Feb 2025 09:44:29 +1000 Subject: [PATCH 10/10] Do not wait for segments down in the test Reason - the downtime it small now, and we can miss the time window. --- gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature | 2 -- 1 file changed, 2 deletions(-) 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