From 1ee85675f0fb0bff9bac908c9e804dba60213bab Mon Sep 17 00:00:00 2001 From: Dianjin Wang Date: Fri, 11 Sep 2026 23:08:25 +0800 Subject: [PATCH 1/7] gprecoverseg: fix the defects the Behave suite exposes Defects the Behave suite exposes in gprecoverseg: * gprecoverseg: point a recovered segment's conf at its own port pg_basebackup and pg_rewind both copy the source segment's postgresql.conf, port setting included, so a mirror recovered by gprecoverseg came back configured for its primary's port. On a multi-host cluster the wrong port is merely wrong; on a single host it collides with the primary that is still running there. Rewrite the port before starting the segment, on both the full and the incremental path, using the existing ModifyConfSetting command -- its perl expression is anchored at the start of the line, so it touches the port setting and nothing else. Greenplum does the same in the same two places. The gpsegrecovery unit tests mock the rewrite, assert it, and expect the line it logs; before, they ran perl on a postgresql.conf that does not exist. Fixes the eight gprecoverseg Behave scenarios that assert "port value in postgresql.conf of is incorrect". * gprecoverseg: keep the progress files a recovery writes Each recovering segment writes pg_basebackup or pg_rewind output to gpAdminLogs/..dbid.out, and gprecoverseg deleted that file as soon as the segment came back successfully. Deleting it on success is backwards: those files are what gpstate reads to report recovery progress, and what an operator reads afterwards to see what a recovery actually did. Remove stale progress files for the segments about to be recovered before starting, and leave the run's own files in place, the way Greenplum does. Clearing beforehand is also what makes the file a reliable record of the current run rather than an append of every run that came before. The stale files are matched by name ("*dbid.out"), so the remove command becomes Greenplum's "find -name -delete". The old "rm -f" form would have quoted the pattern and run it in the remote user's home directory, removing nothing. Its unit tests now assert the single recovery call and the find commands. Fixes the Behave scenarios asserting gpAdminLogs holds "pg_basebackup*" and "pg_rewind*" files after a recovery. * gprecoverseg: reject -F combined with -r or -p A full resynchronisation rebuilds a mirror from its primary in place. That is not what -r asks for -- rebalance returns segments to their preferred roles -- nor what -p asks for, which is to rebuild onto a different host. gprecoverseg accepted both combinations and silently did something other than what was asked. Reject them, with the messages Greenplum uses. * gprecoverseg: report the failure and the progress files it leaves Two log lines Greenplum emits around recover_mirrors() were dropped: - the [ERROR] 'gprecoverseg failed. Please check the output for more details.' that follows a failed recovery. Without it a failed run ends on the per-segment errors alone, and callers that scrape the output -- gpmovemirrors among them -- cannot tell a failure from a partial success. - the notice that the pg_basebackup/pg_rewind progress files this run wrote are left behind and a later gprecoverseg may remove them. Asserted by gpmovemirrors.feature:136 and gprecoverseg.feature:58. * gprecoverseg: reject -o combined with -i or -r -o writes out a sample recovery configuration describing what would be recovered and recovers nothing. -i names a configuration to recover from and -r rebalances back to preferred roles, so either one together with -o is a contradiction: the user asked for a file and for a recovery at the same time, and gprecoverseg quietly honours one of them. Upstream rejects both combinations; the checks live in a method whose other validations are for options this tree does not have, so take just these two and put them beside the "Only one of -i, -p, and -r" check they belong with. They run before the cluster is contacted, so the error is immediate. gprecoverseg.feature has been asserting this and failing. * gprecoverseg: say why a segment failed to recover, not just where When a recovery fails, the summary names the segment and the log file and stops there: Failed to recover the following segments. You must run gprecoverseg -F for all incremental failures hostname: sdw1; port: 7000; logfile: ~/gpAdminLogs/pg_rewind...out; recoverytype: incremental so the one thing the reader wants -- what went wrong -- is the one thing they have to go and find. Upstream ends that line with "; error: {}" and fills it by reading the log back, and the printer for setup-recovery failures in this same file already does the equivalent. Only this one lost it. Bring over get_error_from_logfile, which takes the last ERROR, PANIC or FATAL line from the segment's log. It deliberately skips the line pg_rewind writes about itself after a failure, which is never the reason for it; upstream skips PostgreSQL 12's wording, and this tree needs its own, because pg_rewind.c has since been split into pg_log_error("postgres single-user mode in target cluster failed"); pg_log_error_detail("Command was: %s", ...); With the self-referential line out of the way what is left is the backend's own complaint, which is what a reader needs: for a mirror whose pg_wal was removed, 'required WAL directory "pg_wal" does not exist'. The behave step that checks this summary matches up to "recoverytype: X" with a regex search, so the added field does not disturb it. Split out of #1639 (originally a3bfe47d5c3). --- .../gppylib/operations/buildMirrorSegments.py | 22 +++++-- .../unit/test_unit_buildMirrorSegments.py | 61 ++++++++----------- .../bin/gppylib/programs/clsRecoverSegment.py | 23 +++++++ gpMgmt/bin/gppylib/recoveryinfo.py | 40 +++++++++++- .../test/unit/test_unit_gpsegrecovery.py | 23 ++++++- gpMgmt/sbin/gpsegrecovery.py | 23 ++++++- 6 files changed, 143 insertions(+), 49 deletions(-) diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 869a37d312b..1c5997833ae 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -240,6 +240,9 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild) + # Remove any existing progress files for segments to be recovered + self.remove_existing_progress_files(recovery_info_by_host) + self._run_setup_recovery(actionName, recovery_info_by_host) backout_map = self._update_config(recovery_info_by_host, gpArray) @@ -293,13 +296,17 @@ def _update_config(self, recovery_info_by_host, gpArray): return backout_map - def _remove_progress_files(self, recovery_info_by_host, recovery_results): + # Remove any existing progress file of segments that will be recovered by + # current gprecoverseg execution. The files written by this run are left in + # place: gpstate reads them to report progress, and they are what an + # operator looks at after a recovery that went wrong. + def remove_existing_progress_files(self, recovery_info_by_host): remove_progress_file_cmds = [] for hostName, recovery_info_list in recovery_info_by_host.items(): for ri in recovery_info_list: - if recovery_results.was_bb_rewind_successful(ri.target_segment_dbid): - remove_progress_file_cmds.append(self._get_remove_cmd(ri.progress_file, hostName)) - self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=False) + remove_progress_file_cmds.append(self._get_remove_cmd("*dbid{}.out".format(ri.target_segment_dbid), + hostName)) + self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=True) def _revert_config_update(self, recovery_results, backout_map): if len(backout_map) == 0: @@ -447,8 +454,12 @@ def _get_progress_cmd(self, progressFile, targetSegmentDbId, targetHostname): remoteHost=targetHostname) return None + # remove_file is a name pattern (e.g. "*dbid2.out"), matched in the log + # directory the progress files are written to. def _get_remove_cmd(self, remove_file, target_host): - return base.Command("remove file", "rm -f {}".format(shlex.quote(remove_file)), ctxt=base.REMOTE, remoteHost=target_host) + return base.Command("remove file", "find {} -name {} -delete".format(shlex.quote(gplog.get_logger_dir()), + shlex.quote(remove_file)), + ctxt=base.REMOTE, remoteHost=target_host) def __runWaitAndCheckWorkerPoolForErrorsAndClear(self, cmds, suppressErrorCheck=False, progressCmds=[]): for cmd in cmds: @@ -486,7 +497,6 @@ def _run_recovery(self, action_name, recovery_info_by_host, gpEnv): recovery_results = RecoveryResult(action_name, completed_recovery_results, self.__logger) recovery_results.print_bb_rewind_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): diff --git a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py index 61a88c21539..0154b83d139 100644 --- a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py @@ -522,6 +522,24 @@ def test_run_setup_recovery_empty_errors(self): self._assert_setup_recovery() + def test_remove_existing_progress_files(self): + self._setup_recovery_mocks() + + recovery_info = OrderedDict() + recovery_info['host1'] = [self.recovery_info1] + recovery_info['host2'] = [self.recovery_info2, self.recovery_info3] + + self.default_build_mirrors_obj.remove_existing_progress_files(recovery_info) + + self.assertEqual([call(ANY, suppressErrorCheck=True)], + self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) + rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][0][0] + self.assertEqual(['host1', 'host2', 'host2'], [cmd.remoteHost for cmd in rm_progress_cmds]) + self.assertEqual(["find /tmp/logdir -name '*dbid2.out' -delete", + "find /tmp/logdir -name '*dbid3.out' -delete", + "find /tmp/logdir -name '*dbid4.out' -delete"], + [cmd.cmdStr for cmd in rm_progress_cmds]) + def test_run_recovery_no_errors(self): self._setup_recovery_mocks() @@ -531,12 +549,11 @@ def test_run_recovery_no_errors(self): self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) seg_recovery_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][0][0] progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][1]['progressCmds'] - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] #TODO fix formatting expected_recovery_cmd_strs = ["""$GPHOME/sbin/gpsegrecovery.py -c '[{"target_datadir": "/datadir2", "target_port": 7001, "target_segment_dbid": 2, "source_hostname": "source_host_for_dbid2", "source_port": 7002, "is_full_recovery": true, "progress_file": "/tmp/progress_file2"}]' -l /tmp/logdir -b 64 --era=dummy_era""", """$GPHOME/sbin/gpsegrecovery.py -c '[{"target_datadir": "/datadir3", "target_port": 7003, "target_segment_dbid": 3, "source_hostname": "source_host_for_dbid3", "source_port": 7004, "is_full_recovery": true, "progress_file": "/tmp/progress_file3"}, {"target_datadir": "/datadir4", "target_port": 7005, "target_segment_dbid": 4, "source_hostname": "source_host_for_dbid4", "source_port": 7006, "is_full_recovery": true, "progress_file": "/tmp/progress_file4"}]' -l /tmp/logdir -b 64 --era=dummy_era"""] @@ -558,13 +575,6 @@ def test_run_recovery_no_errors(self): progress_cmds[2].cmdStr) self.assertEqual(1, self.mock_gp_era.call_count) - self.assertEqual(3, len(rm_progress_cmds)) - self.assertEqual("host1", rm_progress_cmds[0].remoteHost) - self.assertEqual("rm -f /tmp/progress_file2", rm_progress_cmds[0].cmdStr) - self.assertEqual("host2", rm_progress_cmds[1].remoteHost) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[1].cmdStr) - self.assertEqual("host2", rm_progress_cmds[2].remoteHost) - self.assertEqual("rm -f /tmp/progress_file4", rm_progress_cmds[2].cmdStr) self._assert_run_recovery() @@ -584,11 +594,9 @@ def test_run_recovery_all_dbids_fail_only_bb_rewind_errors(self): self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(0, len(rm_progress_cmds)) self._assert_run_recovery() @@ -599,13 +607,9 @@ def test_run_recovery_some_dbids_fail_all_bb_rewind_errors(self): self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(1, len(rm_progress_cmds)) - self.assertEqual("host2", rm_progress_cmds[0].remoteHost) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[0].cmdStr) self._assert_run_recovery() @@ -616,11 +620,8 @@ def test_run_recovery_all_dbids_fail_all_start_errors(self): host2_error = '[{"error_type": "start", "error_msg":"some error for dbid 4", "dbid": 4, "datadir": "/datadir4", "port": 7005, "progress_file": "/tmp/progress4"}]' self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - - self.assertEqual(3, len(rm_cmds)) self._assert_run_recovery() def test_run_recovery_some_dbids_fail_all_start_errors(self): @@ -631,10 +632,8 @@ def test_run_recovery_some_dbids_fail_all_start_errors(self): self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(3, len(rm_progress_cmds)) self._assert_run_recovery() @@ -647,14 +646,10 @@ def test_run_recovery_all_dbids_fail_bb_rewind_and_start_errors(self): self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] # Since start passed for dbid:3, we should have deleted it's progress file - self.assertEqual(1, len(rm_progress_cmds)) - self.assertEqual("host1", rm_progress_cmds[0].remoteHost) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[0].cmdStr) self._assert_run_recovery() @@ -671,15 +666,9 @@ def test_run_recovery_all_dbids_fail_bb_rewind_and_default_errors(self): self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(2, len(rm_progress_cmds)) - self.assertEqual("host1", rm_progress_cmds[0].remoteHost) - self.assertEqual("host2", rm_progress_cmds[1].remoteHost) - self.assertEqual("rm -f /tmp/progress_file2", rm_progress_cmds[0].cmdStr) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[1].cmdStr) self._assert_run_recovery() diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py index 81377136f83..503360368a4 100644 --- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py +++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py @@ -262,6 +262,25 @@ def run(self): if optionCnt > 1: raise ProgramArgumentValidationException("Only one of -i, -p, and -r may be specified") + # -o only writes out a sample configuration file describing what would + # be recovered; it does not recover anything, so combining it with an + # option that says where to recover to is a contradiction rather than a + # refinement. + if self.__options.outputSampleConfigFile: + if self.__options.recoveryConfigFile is not None: + raise ProgramArgumentValidationException("Invalid -i provided with -o argument") + if self.__options.rebalanceSegments: + raise ProgramArgumentValidationException("Invalid -r provided with -o argument") + + # A full resynchronisation rebuilds the mirror from its primary, which + # is neither what -r (rebalance back to preferred roles) nor -p + # (recover onto a different host) asks for. + if self.__options.forceFullResynchronization: + if self.__options.rebalanceSegments: + raise ProgramArgumentValidationException("-F option is not supported with -r option") + if self.__options.newRecoverHosts is not None: + raise ProgramArgumentValidationException("-F option is not supported with -p option") + faultProberInterface.getFaultProber().initializeProber(gpEnv.getCoordinatorPort()) confProvider = configInterface.getConfigurationProvider().initializeProvider(gpEnv.getCoordinatorPort()) @@ -378,12 +397,16 @@ def signal_handler(sig, frame): if not mirrorBuilder.recover_mirrors(gpEnv, gpArray): if self.termination_requested: self.logger.error("gprecoverseg process was interrupted by the user.") + self.logger.error("gprecoverseg failed. Please check the output for more details.") sys.exit(1) if self.termination_requested: self.logger.info("Not able to terminate the recovery process since it has been completed successfully.") self.logger.info("********************************") + self.logger.info("Future gprecoverseg executions might remove the currently created pg_basebackup/pg_rewind/rsync " + "progress files, please save these files if needed.") + self.logger.info("********************************") self.logger.info("Segments successfully recovered.") self.logger.info("********************************") diff --git a/gpMgmt/bin/gppylib/recoveryinfo.py b/gpMgmt/bin/gppylib/recoveryinfo.py index f13ea81394b..b4530bc2194 100644 --- a/gpMgmt/bin/gppylib/recoveryinfo.py +++ b/gpMgmt/bin/gppylib/recoveryinfo.py @@ -1,8 +1,10 @@ from collections import defaultdict import datetime import json +import shlex from gppylib import gplog +from gppylib.commands.base import Command, REMOTE class RecoveryInfo(object): @@ -169,8 +171,36 @@ def print_setup_recovery_errors(self): self._logger.error(setup_recovery_error_pattern.format(hostname, error.port, error.error_msg)) self._print_invalid_errors() + # Cherry-picked from Greenplum Database (greenplum-db/gpdb-archive @ 482967c1b4), + # gpMgmt/bin/gppylib/recoveryinfo.py, with the line it filters out + # changed: pg_rewind reports its own failure differently since + # PostgreSQL 13. See the comment in the body. + def get_error_from_logfile(self, progress_file, hostname): + """ + Traverse the error file to fetch last registered occurrence of 'error, panic or fatal'. + pg_rewind logs a line of its own after every failed attempt, which is + never the reason for the failure -- the reason is whatever the backend + wrote just above it -- so that line is excluded. Upstream excludes + PostgreSQL 12's wording of it, "fatal: postgres single-user mode of + target instance failed for command"; since 13 pg_rewind writes + "error: postgres single-user mode in target cluster failed" followed by + a separate "detail: Command was: ...", and only the first of those two + carries a word this grep looks for. + To ensure grep doesn't return a non-zero exit code, it is ORed with true. + """ + cmdStr = r'set -o pipefail; cat {} | (grep -i "ERROR\|PANIC\|FATAL" | grep -v "error: postgres single-user mode in target cluster failed" || true) | tail -1'.format( + shlex.quote(progress_file)) + cmd = Command(name="Parse logfile for errors on a remote host", cmdStr=cmdStr, ctxt=REMOTE, remoteHost=hostname) + cmd.run() + + if cmd.get_results().rc != 0: + self._logger.debug("Failed while parsing the logfile %s" % progress_file) + return None + + return cmd.get_results().stdout.strip() + def print_bb_rewind_and_start_errors(self): - bb_rewind_error_pattern = " hostname: {}; port: {}; logfile: {}; recoverytype: {}" + bb_rewind_error_pattern = " hostname: {}; port: {}; logfile: {}; recoverytype: {}; error: {}" if len(self._bb_errors) > 0 or len(self._rewind_errors) > 0: self._logger.info("----------------------------------------------------------") if len(self._rewind_errors) > 0: @@ -181,11 +211,15 @@ def print_bb_rewind_and_start_errors(self): for hostname, errors in self._rewind_errors.items(): for error in errors: self._logger.info(bb_rewind_error_pattern.format(hostname, error.port, error.progress_file, - error.error_type)) + error.error_type, + self.get_error_from_logfile(error.progress_file, + hostname))) for hostname, errors in self._bb_errors.items(): for error in errors: self._logger.info(bb_rewind_error_pattern.format(hostname, error.port, error.progress_file, - error.error_type)) + error.error_type, + self.get_error_from_logfile(error.progress_file, + hostname))) start_error_pattern = " hostname: {}; port: {}; datadir: {}" if len(self._start_errors) > 0: diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py index 983b5f25a9a..bd380107c7d 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py @@ -18,11 +18,13 @@ def setUp(self): self.mock_logger = Mock() self.apply_patches([ patch('gpsegrecovery.start_segment', return_value=Mock()), + patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()), patch('gppylib.commands.pg.PgRewind.__init__', return_value=None), patch('gppylib.commands.pg.PgRewind.run') ]) self.mock_pgrewind_run = self.get_mock_from_apply_patch('run') self.mock_pgrewind_init = self.get_mock_from_apply_patch('__init__') + self.mock_modify_conf = self.get_mock_from_apply_patch('ModifyConfSetting') p = Segment.initFromString("1|0|p|p|s|u|sdw1|sdw1|40000|/data/primary0") m = Segment.initFromString("2|0|m|m|s|u|sdw2|sdw2|50000|/data/mirror0") @@ -56,8 +58,13 @@ def test_incremental_run_passes(self): self.assertEqual(1, self.mock_pgrewind_run.call_count) self.assertEqual(call(validateAfter=True), self.mock_pgrewind_run.call_args) logger_call_args = [call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file'), - call('Successfully ran pg_rewind for dbid: 2')] + call('Successfully ran pg_rewind for dbid: 2'), + call('Updating /data/mirror0/postgresql.conf')] self.assertEqual(logger_call_args, self.mock_logger.info.call_args_list) + self.mock_modify_conf.assert_called_once_with('Updating /data/mirror0/postgresql.conf', + '/data/mirror0/postgresql.conf', 'port', 50000, + optType='number') + self.mock_modify_conf.return_value.run.assert_called_once_with(validateAfter=True) gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era) def test_incremental_run_exception(self): @@ -71,6 +78,7 @@ def test_incremental_run_exception(self): self.assertEqual(call(validateAfter=True), self.mock_pgrewind_run.call_args) self.assertEqual([call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file')], self.mock_logger.info.call_args_list) + self.assertEqual(0, self.mock_modify_conf.call_count) self.assertEqual(0, gpsegrecovery.start_segment.call_count) self._assert_cmd_failed('{"error_type": "incremental", "error_msg": "pg_rewind failed", "dbid": 2, ' \ '"datadir": "/data/mirror0", "port": 50000, "progress_file": "/tmp/test_progress_file"}') @@ -92,7 +100,8 @@ def test_incremental_start_segment_exception(self): self.assertEqual(1, self.mock_pgrewind_init.call_count) self.assertEqual(1, self.mock_pgrewind_run.call_count) logger_call_args = [call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file'), - call('Successfully ran pg_rewind for dbid: 2')] + call('Successfully ran pg_rewind for dbid: 2'), + call('Updating /data/mirror0/postgresql.conf')] self.assertEqual(logger_call_args, self.mock_logger.info.call_args_list) gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era) self._assert_cmd_failed('{"error_type": "start", "error_msg": "pg_ctl start failed", "dbid": 2, ' \ @@ -107,11 +116,13 @@ def setUp(self): self.mock_logger = Mock(spec=['log', 'info', 'debug', 'error', 'warn', 'exception']) self.apply_patches([ patch('gpsegrecovery.start_segment', return_value=Mock()), + patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()), patch('gpsegrecovery.PgBaseBackup.__init__', return_value=None), patch('gpsegrecovery.PgBaseBackup.run') ]) self.mock_pgbasebackup_run = self.get_mock_from_apply_patch('run') self.mock_pgbasebackup_init = self.get_mock_from_apply_patch('__init__') + self.mock_modify_conf = self.get_mock_from_apply_patch('ModifyConfSetting') p = Segment.initFromString("1|0|p|p|s|u|sdw1|sdw1|40000|/data/primary0") m = Segment.initFromString("2|0|m|m|s|u|sdw2|sdw2|50000|/data/mirror0") @@ -135,8 +146,13 @@ def _assert_basebackup_runs(self, expected_init_args): self.assertEqual(1, self.mock_pgbasebackup_run.call_count) self.assertEqual(call(validateAfter=True), self.mock_pgbasebackup_run.call_args) expected_logger_info_args = [call('Running pg_basebackup with progress output temporarily in /tmp/test_progress_file'), - call("Successfully ran pg_basebackup for dbid: 2")] + call("Successfully ran pg_basebackup for dbid: 2"), + call('Updating /data/mirror0/postgresql.conf')] self.assertEqual(expected_logger_info_args, self.mock_logger.info.call_args_list) + self.mock_modify_conf.assert_called_once_with('Updating /data/mirror0/postgresql.conf', + '/data/mirror0/postgresql.conf', 'port', 50000, + optType='number') + self.mock_modify_conf.return_value.run.assert_called_once_with(validateAfter=True) gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era) def _assert_cmd_passed(self): @@ -277,6 +293,7 @@ def setUp(self): self.apply_patches([ patch('gpsegrecovery.SegmentStart.__init__', return_value=None), patch('gpsegrecovery.SegmentStart.run'), + patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()), ]) self.mock_segment_start_init = self.get_mock_from_apply_patch('__init__') diff --git a/gpMgmt/sbin/gpsegrecovery.py b/gpMgmt/sbin/gpsegrecovery.py index 811a73ccbb5..a1a6891ee35 100644 --- a/gpMgmt/sbin/gpsegrecovery.py +++ b/gpMgmt/sbin/gpsegrecovery.py @@ -7,7 +7,7 @@ from gppylib.commands.pg import PgBaseBackup, PgRewind from recovery_base import RecoveryBase, set_recovery_cmd_results from gppylib.commands.base import Command -from gppylib.commands.gp import SegmentStart +from gppylib.commands.gp import SegmentStart, ModifyConfSetting from gppylib.gparray import Segment from gppylib.commands.unix import terminate_proc_tree @@ -63,6 +63,10 @@ def run(self): self.error_type = RecoveryErrorType.DEFAULT_ERROR self.logger.info("Successfully ran pg_basebackup for dbid: {}".format( self.recovery_info.target_segment_dbid)) + + # Updating port number on conf after recovery + update_port_in_conf(self.recovery_info, self.logger) + self.error_type = RecoveryErrorType.START_ERROR start_segment(self.recovery_info, self.logger, self.era) @@ -87,10 +91,27 @@ def run(self): cmd.run(validateAfter=True) self.logger.info("Successfully ran pg_rewind for dbid: {}".format(self.recovery_info.target_segment_dbid)) + # Updating port number on conf after recovery + update_port_in_conf(self.recovery_info, self.logger) + self.error_type = RecoveryErrorType.START_ERROR start_segment(self.recovery_info, self.logger, self.era) +def update_port_in_conf(recovery_info, logger): + """ + pg_basebackup and pg_rewind both copy the source segment's + postgresql.conf, which carries the *source* port. Point it back at this + segment before it is started, or the mirror comes up on its primary's + port. + """ + 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'), + 'port', recovery_info.target_port, optType='number') + modifyConfCmd.run(validateAfter=True) + + def start_segment(recovery_info, logger, era): seg = Segment(None, None, None, None, None, None, None, None, recovery_info.target_port, recovery_info.target_datadir) From 3814f9b62d99ae2c09ea04156ce9580f7ee2b61f Mon Sep 17 00:00:00 2001 From: Dianjin Wang Date: Thu, 17 Sep 2026 16:37:38 +0800 Subject: [PATCH 2/7] gpstate: stop reporting recovery progress for a gprecoverseg that died gpstate -e decides a recovery is in progress by asking whether $COORDINATOR_DATA_DIRECTORY/gprecoverseg.lock exists. gprecoverseg removes that directory when it exits normally, but a process killed with SIGKILL never gets the chance, so from then on every gpstate -e reports Segments in recovery for a recovery that is not happening, and exits 1 while doing so. Upstream asks a different question: it reads the PID out of the lock directory and checks whether that process is still alive. Bring over is_gprecoverseg_running and use it, with the two unit tests that cover the pid-is-alive and no-pidfile cases. Found by gprecoverseg.feature:735 "SIGKILL on gprecoverseg should not display progress in gpstate -e" on a five-node cluster, where it was the first of 38 failures -- the other 37 were scenarios that inherited the half-recovered cluster it left behind. The get_coordinatordatadir import existed only for the directory check and is now unused, as it is upstream. Split out of #1639 (originally 879b5638026). --- gpMgmt/bin/gppylib/commands/gp.py | 12 ++++++++++ .../commands/test/unit/test_unit_gp.py | 21 ++++++++++++++++-- gpMgmt/bin/gppylib/programs/clsSystemState.py | 22 ++++++++++--------- 3 files changed, 43 insertions(+), 12 deletions(-) diff --git a/gpMgmt/bin/gppylib/commands/gp.py b/gpMgmt/bin/gppylib/commands/gp.py index b980bc629c4..03f6c1269e7 100644 --- a/gpMgmt/bin/gppylib/commands/gp.py +++ b/gpMgmt/bin/gppylib/commands/gp.py @@ -1670,6 +1670,18 @@ def createTempDirectoryName(coordinatorDataDirectory, tempDirPrefix): datetime.datetime.now().strftime('%m%d%Y'), os.getpid()) +# Cherry-picked from Greenplum Database (greenplum-db/gpdb-archive @ 482967c1b4), +# gpMgmt/bin/gppylib/commands/gp.py. +def is_gprecoverseg_running(): + gprecoverseg_pidfile = os.path.join(get_coordinatordatadir(), 'gprecoverseg.lock', 'PID') + try: + with open(gprecoverseg_pidfile) as pidfile: + gprecoverseg_pid = pidfile.read() + except Exception: + return False + + return check_pid(gprecoverseg_pid) + #------------------------------------------------------------------------- class GpRecoverSeg(Command): """ diff --git a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py index 4ac75c69685..018641a6682 100644 --- a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py +++ b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py @@ -4,9 +4,9 @@ # from gppylib.commands.base import CommandResult -from mock import patch +from mock import patch, mock_open -from gppylib.commands.gp import is_pid_postmaster, get_postmaster_pid_locally +from gppylib.commands.gp import is_pid_postmaster, get_postmaster_pid_locally, is_gprecoverseg_running from test.unit.gp_unittest import GpTestCase, run_tests @@ -158,5 +158,22 @@ def test_get_postmaster_pid_locally_error(self, mock1): def test_get_postmaster_pid_locally_empty(self, mock1): self.assertEqual(get_postmaster_pid_locally('/tmp'), -1) + @patch('gppylib.commands.gp.check_pid', return_value=True) + @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/data/coordinator') + @patch("builtins.open", new_callable=mock_open, read_data="123") + def test_is_gprecoverseg_running_succeeds(self, mock_file, mock1, mock2): + result = is_gprecoverseg_running() + mock_file.assert_called_once_with('/data/coordinator/gprecoverseg.lock/PID') + mock2.assert_called_once_with('123') + self.assertTrue(result) + + @patch('gppylib.commands.gp.check_pid') + @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/invalid/path/') + def test_is_gprecoverseg_running_when_pidfile_does_not_exists(self, mock1, mock2): + result = is_gprecoverseg_running() + self.assertFalse(result) + self.assertFalse(mock2.called) + + if __name__ == '__main__': run_tests() diff --git a/gpMgmt/bin/gppylib/programs/clsSystemState.py b/gpMgmt/bin/gppylib/programs/clsSystemState.py index e234b633122..2434adb955d 100644 --- a/gpMgmt/bin/gppylib/programs/clsSystemState.py +++ b/gpMgmt/bin/gppylib/programs/clsSystemState.py @@ -22,7 +22,6 @@ from gppylib.system import configurationInterface as configInterface from gppylib.system.environment import GpCoordinatorEnvironment from gppylib.utils import TableLogger -from gppylib.commands.gp import get_coordinatordatadir logger = gplog.get_default_logger() @@ -682,15 +681,18 @@ def logSegments(segments, logAsPairs, additionalFieldsToLog=[]): self.__addClusterDownWarning(gpArray, data) - recovery_progress_file = get_recovery_progress_file(gplog) - segments_under_recovery = self._parse_recovery_progress_data(data, recovery_progress_file, gpArray) - gprecoverseg_lock_dir = os.path.join(get_coordinatordatadir() + '/gprecoverseg.lock') - if segments_under_recovery and os.path.exists(gprecoverseg_lock_dir): - logger.info("----------------------------------------------------") - logger.info("Segments in recovery") - logSegments(segments_under_recovery, False, [VALUE_RECOVERY_TYPE, VALUE_RECOVERY_COMPLETED_BYTES, VALUE_RECOVERY_TOTAL_BYTES, - VALUE_RECOVERY_PERCENTAGE]) - exitCode = 1 + # Check if gprecoverseg process is running. We do not want to + # show the progress if the gprecoverseg process is not running. + if gp.is_gprecoverseg_running(): + recovery_progress_file = get_recovery_progress_file(gplog) + segments_under_recovery = self._parse_recovery_progress_data(data, recovery_progress_file, gpArray) + + if segments_under_recovery: + logger.info("----------------------------------------------------") + logger.info("Segments in recovery") + logSegments(segments_under_recovery, False, [VALUE_RECOVERY_TYPE, VALUE_RECOVERY_COMPLETED_BYTES, + VALUE_RECOVERY_TOTAL_BYTES, VALUE_RECOVERY_PERCENTAGE]) + exitCode = 1 # final output -- no errors, then log this message if exitCode == 0: From 5ebd7d797b75859c4549164a52d8e4f3e119fa9d Mon Sep 17 00:00:00 2001 From: Dianjin Wang Date: Sat, 10 Oct 2026 11:29:12 +0800 Subject: [PATCH 3/7] gprecoverseg: skip segments that already have a recovery running There must be one writer per target segment. If a pg_basebackup or a pg_rewind is already streaming into a segment and gprecoverseg starts a second one, the two races leave the segment inconsistent. Greenplum guards against this by asking gp_stat_replication which segments are already sourcing a pg_basebackup, and pg_stat_activity whether a pg_rewind is connected, then skipping those and saying so. Cloudberry carried over neither check, so a second recovery would just start. Port both. gpmovemirrors needs no change of its own: it moves mirrors by shelling out to gprecoverseg, which is why its three scenarios assert on what gprecoverseg printed. This lets seven scenarios run instead of being skipped -- four in gprecoverseg.feature and three in gpmovemirrors.feature, all of the form "gives warning if pg_basebackup already running for ... segments". cherry-pick from Greenplum Database, greenplum-db/gpdb-archive @ 482967c1b4, gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py. One deliberate departure from upstream: Greenplum skips a third case, a source segment already in backup mode, which only arises from its differential recovery. Cloudberry has no differential recovery, so that arm and its pg_is_in_backup() dependency are left out. Split out of #1639 (originally 7056e02e306). --- .../programs/clsRecoverSegment_triples.py | 113 +++++++++++++++++- 1 file changed, 112 insertions(+), 1 deletion(-) diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py index 34b6a12ff2e..5f2cd466a5b 100644 --- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py +++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py @@ -1,11 +1,76 @@ import abc from typing import List +from contextlib import closing +from gppylib.db import dbconn +from gppylib import gplog from gppylib.mainUtils import ExceptionNoStackTraceNeeded from gppylib.operations.detect_unreachable_hosts import get_unreachable_segment_hosts from gppylib.parseutils import line_reader, check_values, canonicalize_address from gppylib.utils import checkNotNone, normalizeAndValidateInputPath from gppylib.gparray import GpArray, Segment +from gppylib.commands.gp import RECOVERY_REWIND_APPNAME + +logger = gplog.get_default_logger() + + +# cherry-pick from Greenplum Database +# greenplum-db/gpdb-archive @ 482967c1b4, +# gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py +def get_segments_with_running_basebackup(): + """ + Returns a list of contentIds of source segments of running pg_basebackup processes + At present gp_stat_replication table does not contain any info about datadir and dbid of the target of running pg_basebackup + """ + + sql = "select gp_segment_id from gp_stat_replication where application_name = 'pg_basebackup'" + + try: + with closing(dbconn.connect(dbconn.DbURL())) as conn: + res = dbconn.query(conn, sql) + rows = res.fetchall() + except Exception as e: + raise Exception("Failed to query gp_stat_replication: %s" %str(e)) + + segments_with_running_basebackup = {row[0] for row in rows} + + if len(segments_with_running_basebackup) == 0: + logger.debug("No basebackup running") + return segments_with_running_basebackup + + +# cherry-pick from Greenplum Database +# greenplum-db/gpdb-archive @ 482967c1b4, +# gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py +def is_pg_rewind_running(hostname, port): + """ + Returns true if a pg_rewind process is running for the given segment + """ + logger.debug( + "Checking for running instances of pg_rewind with host {} and port {} as source server".format(hostname, port)) + + """ + Reasons to depend on pg_stat_activity table: + * pg_rewind is invoked using --source-server connection string as it needs libpq connection + with source server, which will be remote to the target server and --source-pgdata can not + be used in that case. Thus pg_stat_activity will always contain entry for active pg_rewind. + * pg_rewind keeps a connection open throughout it's lifecycle, so pg_stat_activity will contain + entries for active pg_rewind process till the end of execution. + * gpstate uses the above mentioned approach (pg_stat_activity entry check) to check for + incremental recoveries in progress.Thus, using the same approach will make it consistent + everywhere. + """ + + sql = "SELECT count(*) FROM pg_stat_activity WHERE application_name = '{}'".format(RECOVERY_REWIND_APPNAME) + try: + url = dbconn.DbURL(hostname=hostname, port=port, dbname='template1') + with closing(dbconn.connect(url, utility=True)) as conn: + res = dbconn.querySingleton(conn, sql) + return res > 0 + + except Exception as e: + raise Exception("Failed to query pg_stat_activity for segment hostname: {}, port: {}, error: {}".format( + hostname, str(port), str(e))) class RecoveryTriplet: @@ -142,7 +207,31 @@ def _convert_requests_to_triplets(self, requests: List[RecoveryTripletRequest]) triplets = [] dbIdToPeerMap = self.gpArray.getDbIdToPeerMap() + + failed_segments_with_running_basebackup = [] + failed_segments_with_running_pgrewind = [] + segments_with_running_basebackup = get_segments_with_running_basebackup() + for req in requests: + """ + When running gprecoverseg (any sort of recovery full/incremental), if the pg_rewind, pg_basebackup is + already running for a segment, that segment should be skipped from the recovery. The reason being that + there should be only one writer per target segment at a time. Having several writers to a target will + eventually make the segment inconsistent and in a weird state. + + Although technically we could allow user to run a full recovery to a new host even if there is a + pg_rewind/pg_basebackup running for that segment. This is a pretty rare scenario and we have decided not + to over complicates the code just to support this scenario. + """ + # The peer has to be looked up before the failover block below, which + # replaces req.failed with a copy; the dbid is the same either way, + # but the lookup is needed here to decide whether to skip at all. + failed_segment_dbid = req.failed.getSegmentDbId() + peer = dbIdToPeerMap.get(failed_segment_dbid) + if peer is None: + raise Exception("No peer found for dbid {}. liveSegment is None".format(failed_segment_dbid)) + peer_contentid = peer.getSegmentContentId() + # TODO: These 2 cases have different behavior which might be confusing to the user. # "|| ||" does full recovery # "||" does incremental recovery @@ -166,10 +255,32 @@ def _convert_requests_to_triplets(self, requests: List[RecoveryTripletRequest]) if req.failed.unreachable and not req.failover_to_new_host: continue - peer = dbIdToPeerMap.get(req.failed.getSegmentDbId()) + # Only now, once this segment is actually going to be recovered. + # is_pg_rewind_running() connects to the peer, and on a double + # fault -- failed segment and peer both down -- that connection + # raises. Asked before the skip above, it would abort the whole + # gprecoverseg run rather than skip the one segment, which is the + # opposite of what is wanted when a cluster is already degraded. + if peer_contentid in segments_with_running_basebackup: + failed_segments_with_running_basebackup.append(peer_contentid) + continue + + if is_pg_rewind_running(peer.getSegmentHostName(), peer.getSegmentPort()): + failed_segments_with_running_pgrewind.append(peer_contentid) + continue triplets.append(RecoveryTriplet(req.failed, peer, failover)) + if len(failed_segments_with_running_basebackup) > 0: + logger.warning( + "Found pg_basebackup running for segments with contentIds %s, skipping recovery of these segments" % ( + failed_segments_with_running_basebackup)) + + if len(failed_segments_with_running_pgrewind) > 0: + logger.warning( + "Found pg_rewind running for segments with contentIds %s, skipping recovery of these segments" % ( + failed_segments_with_running_pgrewind)) + return triplets From d7665a616f8012dd3337f19bb2472d52ea54c80d Mon Sep 17 00:00:00 2001 From: Dianjin Wang Date: Sat, 10 Oct 2026 11:33:48 +0800 Subject: [PATCH 4/7] gprecoverseg: accept a hostname in the -i recovery config file A segment has both a hostname and an address, and they are not always the same -- the address is whichever interface the segment listens on, and a host with several interfaces has several of them. gprecoverseg's -i config file only understood "address|port|datadir", so on such a cluster it had to guess the hostname from the address, and a recovery to a new host could register the segment under the wrong one. Greenplum lets the line carry the hostname explicitly, as "hostname|address|port|datadir", and when it does: - the hostname is cross-checked against the address, by resolving both and requiring a common IP, so a typo is caught before anything is recovered rather than after; - the failed segment is matched on hostname as well, so a line naming the wrong host no longer silently recovers the right one; - the error naming an unfound segment quotes all four fields. Port that. Both groups on a line must use the same number of fields, which is what the "expected equal parts" error says. cherry-pick from Greenplum Database, greenplum-db/gpdb-archive @ 482967c1b4: gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py gpMgmt/bin/gppylib/utils.py This lets nine scenarios run instead of being skipped, the "populates hostname and address from the config file correctly" group and the two that assert the cross-check warning. Two deliberate departures from upstream. Greenplum also accepts a fifth field holding a per-line recovery type (I/D/F), which exists for its differential recovery; Cloudberry has no differential recovery and no recovery-type field, so only the 3- and 4-part forms are accepted and the error message says so. And validateHostnameAddress() calls socket.gethostbyaddr() rather than importing the bare name, because gppylib/utils.py already imports the socket module. One further departure is a fix rather than a trim. Upstream resolves both names with gethostbyaddr() and requires their address lists to intersect. gethostbyaddr() answers for a single address family, so on a dual-stack host "localhost" comes back as ::1 while an address naming that same machine comes back as 127.0.0.1; the lists do not intersect and the check reports two different hosts. The consequence is not a failure but a silence -- the segment is skipped and gprecoverseg still exits 0, having recovered nothing. Resolve through getaddrinfo() instead and take every address of each name, so the comparison no longer depends on which family the resolver happens to answer with first. Observed on the box this was tested on: localhost -> 127.0.0.1, ::1 cdw -> 127.0.0.1 which upstream's check calls a mismatch. Greenplum also reworked how an unreachable failover host is detected while it was here, probing the hosts in parallel through a new _get_unreachable_failover_hosts() and threading paralleldegree and outputConfigFile down for it. That is a separate concern from reading a hostname out of the config file, and Cloudberry already has its own unreachable handling that the suite exercises, so this keeps that and leaves the rework alone. The -o sample config file still writes three fields, as upstream does. Split out of #1639 (originally 0aa26662c2c). --- .../programs/clsRecoverSegment_triples.py | 114 ++++++++++++++---- gpMgmt/bin/gppylib/utils.py | 42 +++++++ 2 files changed, 135 insertions(+), 21 deletions(-) diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py index 5f2cd466a5b..674387f9807 100644 --- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py +++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py @@ -7,7 +7,7 @@ from gppylib.mainUtils import ExceptionNoStackTraceNeeded from gppylib.operations.detect_unreachable_hosts import get_unreachable_segment_hosts from gppylib.parseutils import line_reader, check_values, canonicalize_address -from gppylib.utils import checkNotNone, normalizeAndValidateInputPath +from gppylib.utils import checkNotNone, normalizeAndValidateInputPath, validateHostnameAddress from gppylib.gparray import GpArray, Segment from gppylib.commands.gp import RECOVERY_REWIND_APPNAME @@ -73,6 +73,29 @@ def is_pg_rewind_running(hostname, port): hostname, str(port), str(e))) +# cherry-pick from Greenplum Database +# greenplum-db/gpdb-archive @ 482967c1b4, +# gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py +def extract_recovery_config_info(parts): + """ + Extracts relevant recovery configuration information from a list of parts. + + Three parts is the historical form, "
||", and the + hostname is taken to be the address. Four parts puts an explicit hostname + first; it is then cross-checked against the address, and against the + hostname the segment is registered under. + """ + address, port, datadir = parts[-3:] + hostname_check_required = False + hostname = address + + if len(parts) == 4: + hostname = parts[0] + hostname_check_required = True + + return hostname, address, port, datadir, hostname_check_required + + class RecoveryTriplet: """ Represents the segments needed to perform a recovery on a given segment. @@ -149,10 +172,12 @@ def validate(failed, live, failover): class RecoveryTripletRequest: - def __init__(self, failed, failover_host=None, failover_port=None, failover_datadir=None, is_new_host=False): + def __init__(self, failed, failover_host_name=None, failover_host_address=None, failover_port=None, + failover_datadir=None, is_new_host=False): self.failed = failed - self.failover_host = failover_host + self.failover_host_name = failover_host_name + self.failover_host_address = failover_host_address self.failover_port = failover_port self.failover_datadir = failover_datadir self.failover_to_new_host = is_new_host @@ -236,7 +261,7 @@ def _convert_requests_to_triplets(self, requests: List[RecoveryTripletRequest]) # "|| ||" does full recovery # "||" does incremental recovery failover = None - if req.failover_host: + if req.failover_host_address: # these two lines make it so that failover points to the object that is registered in gparray # as the failed segment(!). @@ -244,8 +269,17 @@ def _convert_requests_to_triplets(self, requests: List[RecoveryTripletRequest]) req.failed = failover.copy() # now update values in failover segment - failover.setSegmentAddress(req.failover_host) - failover.setSegmentHostName(req.failover_host) + if req.failover_host_name != req.failover_host_address: + # Validate if the hostname and address are of the same host + if not validateHostnameAddress(req.failover_host_name, req.failover_host_address): + logger.warning( + "Not able to co-relate hostname:{0} with address:{1}. " + "Skipping recovery for segments with contentId {2}" + .format(req.failover_host_name, req.failover_host_address, peer_contentid)) + continue + + failover.setSegmentHostName(req.failover_host_name) + failover.setSegmentAddress(req.failover_host_address) failover.setSegmentPort(int(req.failover_port)) failover.setSegmentDataDirectory(req.failover_datadir) failover.unreachable = False if req.failover_to_new_host else failover.unreachable @@ -333,7 +367,11 @@ def _check_new_hosts(): for failedHost, failoverHost in zip(sorted(failedSegments.keys()), self.newHosts): for failed in failedSegments[failedHost]: failoverPort = self.portAssigner.findAndReservePort(failoverHost, failoverHost) - req = RecoveryTripletRequest(failed, failoverHost, failoverPort, failed.getSegmentDataDirectory(), True) + req = RecoveryTripletRequest(failed, failover_host_name=failoverHost, + failover_host_address=failoverHost, + failover_port=failoverPort, + failover_datadir=failed.getSegmentDataDirectory(), + is_new_host=True) requests.append(req) return self._convert_requests_to_triplets(requests) @@ -396,6 +434,11 @@ def getTriplets(self): def _find_failed_from_row(): failed = None for segment in self.gpArray.getDbList(): + # When the input configuration file carries an explicit hostname + # it has to match the one the segment is registered under, or + # this is not the segment the operator meant. + if row['hostname_check_required'] and segment.getSegmentHostName() != row['failedHostname']: + continue if (segment.getSegmentAddress() == row['failedAddress'] and str(segment.getSegmentPort()) == row['failedPort'] and segment.getSegmentDataDirectory() == row['failedDataDirectory']): @@ -403,15 +446,32 @@ def _find_failed_from_row(): break if failed is None: - raise Exception("A segment to recover was not found in configuration. " \ - "This segment is described by address|port|directory '%s|%s|%s'" % - (row['failedAddress'], row['failedPort'], row['failedDataDirectory'])) + # Keep Cloudberry's wording -- the two spaces and the names of + # the fields -- rather than upstream's terser form. It tells + # the operator which field is which, and the scenario that + # covers this asserts on it literally. + msg = "A segment to recover was not found in configuration. " \ + "This segment is described by " + if row['hostname_check_required']: + msg += "hostname|address|port|directory '{}|{}|{}|{}'".format( + row['failedHostname'], row['failedAddress'], + row['failedPort'], row['failedDataDirectory']) + else: + msg += "address|port|directory '{}|{}|{}'".format( + row['failedAddress'], row['failedPort'], + row['failedDataDirectory']) + + raise Exception(msg) return failed requests = [] for row in self.rows: - req = RecoveryTripletRequest(_find_failed_from_row(), row.get('newAddress'), row.get('newPort'), row.get('newDataDirectory')) + req = RecoveryTripletRequest(_find_failed_from_row(), + failover_host_name=row.get('newHostname'), + failover_host_address=row.get('newAddress'), + failover_port=row.get('newPort'), + failover_datadir=row.get('newDataDirectory')) requests.append(req) return self._convert_requests_to_triplets(requests) @@ -433,31 +493,43 @@ def _parseConfigFile(config_file): msg = "line %d of file %s: expected 1 or 2 groups but found %d" % (lineno, config_file, len(groups)) raise ExceptionNoStackTraceNeeded(msg) parts = groups[0].split('|') - if len(parts) != 3: - msg = "line %d of file %s: expected 3 parts on failed segment group, obtained %d" % ( + + if len(parts) not in {3, 4}: + msg = "line {0} of file {1}: expected 3 or 4 parts on failed segment group, obtained {2}" .format( lineno, config_file, len(parts)) raise ExceptionNoStackTraceNeeded(msg) - address, port, datadir = parts - check_values(lineno, address=address, port=port, datadir=datadir) + + hostname, address, port, datadir, hostname_check_required = extract_recovery_config_info(parts) + + check_values(lineno, hostname=hostname, address=address, port=port, datadir=datadir) datadir = normalizeAndValidateInputPath(datadir, f.name, lineno) row = { + 'failedHostname': hostname, 'failedAddress': address, 'failedPort': port, 'failedDataDirectory': datadir, - 'lineno': lineno + 'lineno': lineno, + 'hostname_check_required': hostname_check_required } if len(groups) == 2: parts2 = groups[1].split('|') - if len(parts2) != 3: - msg = "line %d of file %s: expected 3 parts on new segment group, obtained %d" % ( - lineno, config_file, len(parts2)) + if len(parts2) not in [3, 4] or len(parts) != len(parts2): + msg = "line {0} of file {1}: expected equal parts, either 3 or 4 on both segment group, obtained {2} on " \ + "group1 and {3} on group2" .format( + lineno, config_file, len(parts), len(parts2)) raise ExceptionNoStackTraceNeeded(msg) - address2, port2, datadir2 = parts2 - check_values(lineno, address=address2, port=port2, datadir=datadir2) + if len(parts2) == 4: + hostname2, address2, port2, datadir2 = parts2 + else: + address2, port2, datadir2 = parts2 + hostname2 = address2 + + check_values(lineno, hostname=hostname2, address=address2, port=port2, datadir=datadir2) datadir2 = normalizeAndValidateInputPath(datadir2, f.name, lineno) row.update({ + 'newHostname': hostname2, 'newAddress': address2, 'newPort': port2, 'newDataDirectory': datadir2 diff --git a/gpMgmt/bin/gppylib/utils.py b/gpMgmt/bin/gppylib/utils.py index 1adb1e68cdc..000371ce0c8 100644 --- a/gpMgmt/bin/gppylib/utils.py +++ b/gpMgmt/bin/gppylib/utils.py @@ -607,3 +607,45 @@ def formatInsertValuesList(row, starelid, inclHLL): rowVals.append('\t{0}::{1}'.format(val, typ)) return rowVals + + +def _resolve_all_addresses(name): + """ + Every address a name resolves to, across address families. + + Not gethostbyaddr(): it answers for one family only, so on a dual-stack + host "localhost" can come back as ::1 while an address that names the same + machine comes back as 127.0.0.1. Comparing those two lists says the names + are different hosts when they are not. + """ + return {info[4][0] for info in socket.getaddrinfo(name, None)} + + +# cherry-pick from Greenplum Database +# greenplum-db/gpdb-archive @ 482967c1b4, gpMgmt/bin/gppylib/utils.py +def validateHostnameAddress(hostname, address): + """ + validateHostnameAddress : validates that given hostname and address are for the same host + Address can be hostname/alias also. Resolves hostname and address both to get the associated addresses. + + @param hostname Name of the host to be validated + @param address Address of the host to be validated. Can be hostname/alias also + @return if the address and hostname are of the same host + """ + try: + resolved_address_list = _resolve_all_addresses(hostname) + resolved_address_list_2 = _resolve_all_addresses(address) + except Exception as e: + # This means given hostname or address is not reachable + logger.warning( + "Could not resolve hostname:{0}." + .format(hostname)) + return False + + # Resolved address and hostname should have at least one IP address common if they are of same host + if not bool(resolved_address_list.intersection(resolved_address_list_2)): + logger.warning( + "Given address:{0} not present in resolved hostname:{1} address list we got:{2}".format( + address, hostname, sorted(resolved_address_list))) + return False + return True From d5fc5deba89988ded4245669893f9f98f0639075 Mon Sep 17 00:00:00 2001 From: Dianjin Wang Date: Sat, 10 Oct 2026 11:20:52 +0800 Subject: [PATCH 5/7] gprecoverseg: port the rebalance replay-lag check from Greenplum Rebalancing promotes the mirror and demotes the primary. If the mirror still has a large amount of WAL left to replay, that promotion stalls until it catches up, with the segment unavailable for the duration. Greenplum guards against this by measuring the replay lag before rebalancing a pair and refusing when it exceeds a limit. Cloudberry carries the file this lives in but not the check: replay_lag(), the block in rebalance() that calls it, the ALLOWED_REPLAY_LAG default and the two options that drive it had all been dropped. Port them back, so these two scenarios can run instead of being skipped: gprecoverseg rebalance aborts and throws exception if replay lag on mirror is more than or equal to the allowed limit gprecoverseg errors out if invalid options are used with --disable-replay-lag cherry-pick from Greenplum Database, greenplum-db/gpdb-archive @ 482967c1b4: gpMgmt/bin/gppylib/operations/rebalanceSegments.py gpMgmt/bin/gppylib/programs/clsRecoverSegment.py gpMgmt/bin/gppylib/commands/gp.py gpMgmt/bin/gppylib/test/unit/test_unit_rebalance_segment.py The one deliberate departure from upstream: Greenplum reaches the same check through a differential-recovery code path that Cloudberry does not have, so only the rebalance caller is ported. Split out of #1639 (originally ad3c9d59b19). --- gpMgmt/bin/gppylib/commands/gp.py | 5 ++ .../gppylib/operations/rebalanceSegments.py | 53 +++++++++++++++- .../bin/gppylib/programs/clsRecoverSegment.py | 12 +++- .../test/unit/test_unit_gprecoverseg.py | 2 + .../test/unit/test_unit_rebalance_segment.py | 62 ++++++++++++++++++- 5 files changed, 130 insertions(+), 4 deletions(-) diff --git a/gpMgmt/bin/gppylib/commands/gp.py b/gpMgmt/bin/gppylib/commands/gp.py index 03f6c1269e7..d5c5646b61c 100644 --- a/gpMgmt/bin/gppylib/commands/gp.py +++ b/gpMgmt/bin/gppylib/commands/gp.py @@ -53,6 +53,11 @@ #max batch size of thread pool on coordinator MAX_COORDINATOR_NUM_WORKERS=64 +# Maximum replay lag (in GBs) allowed on mirror when rebalancing the segments +# The default value for ALLOWED_REPLAY_LAG has been decided to be 10 GBs as mirror +# took 5 mins to replay 10 GB lag on a local demo cluster. +ALLOWED_REPLAY_LAG = 10 + # Application name used by the pg_rewind instance that gprecoverseg starts # during incremental recovery. gpstate uses this to figure out when incremental # recovery is active. diff --git a/gpMgmt/bin/gppylib/operations/rebalanceSegments.py b/gpMgmt/bin/gppylib/operations/rebalanceSegments.py index a46f0d69d13..a87fdd81340 100644 --- a/gpMgmt/bin/gppylib/operations/rebalanceSegments.py +++ b/gpMgmt/bin/gppylib/operations/rebalanceSegments.py @@ -1,5 +1,6 @@ import sys import signal +from contextlib import closing from gppylib.gparray import GpArray from gppylib.db import dbconn from gppylib.commands.gp import GpSegStopCmd @@ -10,6 +11,43 @@ MIRROR_PROMOTION_TIMEOUT=600 +logger = gplog.get_default_logger() + + +# cherry-pick from Greenplum Database +# greenplum-db/gpdb-archive @ 482967c1b4, +# gpMgmt/bin/gppylib/operations/rebalanceSegments.py +def replay_lag(primary_db): + """ + This function returns replay lag (diff of flush_lsn and replay_lsn) on mirror segment. Goal being if there is a + lot to catchup on mirror the user should be warned about that and rebalance opertion should be aborted. + params: primary segment info + return value: replay lag in bytes + replay lag in bytes: diff of flush_lsn and replay_lsn on mirror + """ + port = primary_db.getSegmentPort() + host = primary_db.getSegmentHostName() + logger.debug('Get replay lag on mirror of primary segment with host:{}, port:{}'.format(host, port)) + sql = "select pg_wal_lsn_diff(flush_lsn, replay_lsn) from pg_stat_replication;" + + try: + dburl = dbconn.DbURL(hostname=host, port=port) + with closing(dbconn.connect(dburl, utility=True, encoding='UTF8')) as conn: + replay_lag = dbconn.querySingleton(conn, sql) + except Exception as ex: + raise Exception("Failed to query pg_stat_replication for host:{}, port:{}, error: {}". + format(host, port, str(ex))) + + # flush_lsn and replay_lsn are nullable -- a walsender that has connected + # but not yet reported leaves pg_wal_lsn_diff() NULL. Upstream hands that + # straight to float(), which dies with a bare TypeError in the middle of a + # rebalance; say what actually happened instead. + if replay_lag is None: + raise Exception("Could not read the replay lag on the mirror of the segment with host:{}, port:{}: " + "the walsender has not reported a flush or replay position yet".format(host, port)) + + return replay_lag + class ReconfigDetectionSQLQueryCommand(base.SQLCommand): """A distributed query that will cause the system to detect @@ -26,11 +64,13 @@ def run(self): class GpSegmentRebalanceOperation: - def __init__(self, gpEnv, gpArray, batch_size, segment_batch_size): + def __init__(self, gpEnv, gpArray, batch_size, segment_batch_size, disable_replay_lag, replay_lag): self.gpEnv = gpEnv self.gpArray = gpArray self.batch_size = batch_size self.segment_batch_size = segment_batch_size + self.disable_replay_lag = disable_replay_lag + self.replay_lag = replay_lag self.logger = gplog.get_default_logger() def rebalance(self): @@ -45,6 +85,17 @@ def rebalance(self): continue if segmentPair.up() and segmentPair.reachable() and segmentPair.synchronized(): + + if not self.disable_replay_lag: + self.logger.info("Allowed replay lag during rebalance is {} GB".format(self.replay_lag)) + replay_lag_in_bytes = replay_lag(segmentPair.primaryDB) + if float(replay_lag_in_bytes) >= (self.replay_lag * 1024 * 1024 * 1024): + raise Exception("{} bytes of wal is still to be replayed on mirror with dbid {}, let " + "mirror catchup on replay then trigger rebalance. Use --replay-lag to " + "configure the allowed replay lag limit or --disable-replay-lag to disable" + " the check completely if you wish to continue with rebalance anyway" + .format(replay_lag_in_bytes, segmentPair.primaryDB.getSegmentDbId())) + unbalanced_primary_segs.append(segmentPair.primaryDB) else: self.logger.warning( diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py index 503360368a4..890ecf855b5 100644 --- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py +++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py @@ -122,7 +122,9 @@ def outputToFile(self, mirrorBuilder, gpArray, fileName): def getRecoveryActionsBasedOnOptions(self, gpEnv, gpArray): if self.__options.rebalanceSegments: - return GpSegmentRebalanceOperation(gpEnv, gpArray, self.__options.parallelDegree, self.__options.parallelPerHost) + return GpSegmentRebalanceOperation(gpEnv, gpArray, self.__options.parallelDegree, + self.__options.parallelPerHost, self.__options.disableReplayLag, + self.__options.replayLag) else: instance = RecoveryTripletsFactory.instance(gpArray, self.__options.recoveryConfigFile, self.__options.newRecoverHosts) segs = [GpMirrorToBuild(t.failed, t.live, t.failover, self.__options.forceFullResynchronization) for t in instance.getTriplets()] @@ -272,6 +274,9 @@ def run(self): if self.__options.rebalanceSegments: raise ProgramArgumentValidationException("Invalid -r provided with -o argument") + if self.__options.disableReplayLag and not self.__options.rebalanceSegments: + raise ProgramArgumentValidationException("--disable-replay-lag should be used only with -r") + # A full resynchronisation rebuilds the mirror from its primary, which # is neither what -r (rebalance back to preferred roles) nor -p # (recover onto a different host) asks for. @@ -523,6 +528,11 @@ def createParser(): addTo.add_option("-r", None, default=False, action='store_true', dest='rebalanceSegments', help='Rebalance synchronized segments.') + addTo.add_option("--replay-lag", None, type="float", default=gp.ALLOWED_REPLAY_LAG, + dest="replayLag", + metavar="", help='Allowed replay lag on mirror, lag should be provided in GBs') + addTo.add_option("--disable-replay-lag", None, default=False, action='store_true', + dest='disableReplayLag', help='Disable replay lag check when rebalancing segments') addTo.add_option('', '--hba-hostnames', action='store_true', dest='hba_hostnames', help='use hostnames instead of CIDR in pg_hba.conf') diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py index 292a6b8b5f1..cdf3f8a21ce 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py @@ -23,6 +23,8 @@ def __init__(self): self.recoveryConfigFile = None self.outputSpareDataDirectoryFile = None self.rebalanceSegments = None + self.disableReplayLag = None + self.replayLag = None self.outputSampleConfigFile = None self.parallelDegree = 1 diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_rebalance_segment.py b/gpMgmt/bin/gppylib/test/unit/test_unit_rebalance_segment.py index 483bb1cb1a8..d9058757632 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_rebalance_segment.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_rebalance_segment.py @@ -4,6 +4,7 @@ from gppylib.gparray import GpArray, Segment from gppylib.commands.base import CommandResult from gppylib.operations.rebalanceSegments import GpSegmentRebalanceOperation +from gppylib.operations.rebalanceSegments import replay_lag class RebalanceSegmentsTestCase(GpTestCase): @@ -11,10 +12,20 @@ def setUp(self): self.pool = Mock() self.pool.getCompletedItems.return_value = [] + mock_logger = Mock(spec=['log', 'warn', 'info', 'debug', 'error', 'warning', 'fatal']) + self.apply_patches([ patch("gppylib.commands.base.WorkerPool.__init__", return_value=None), patch("gppylib.commands.base.WorkerPool", return_value=self.pool), patch('gppylib.programs.clsRecoverSegment.GpRecoverSegmentProgram'), + patch('gppylib.operations.rebalanceSegments.logger', return_value=mock_logger), + patch('gppylib.db.dbconn.connect', autospec=True), + patch('gppylib.db.dbconn.querySingleton', return_value='5678'), + # Upstream only mocks dbconn.connect, which is enough there because + # its _trigger_fts_probe() goes through dbconn. Cloudberry's calls + # pg.connect() directly, so without this every test that lets + # rebalance() run to completion tries to open a real connection. + patch('gppylib.operations.segment_reconfigurer.pg.connect'), ]) self.mock_gp_recover_segment_prog_class = self.get_mock_from_apply_patch('GpRecoverSegmentProgram') @@ -32,8 +43,11 @@ def setUp(self): self.success_command_mock.get_results.return_value = CommandResult( 0, b"stdout success text", b"stderr text", True, False) - self.subject = GpSegmentRebalanceOperation(Mock(), self._create_gparray_with_2_primary_2_mirrors(), 1, 1) - self.subject.logger = Mock() + self.subject = GpSegmentRebalanceOperation(Mock(), self._create_gparray_with_2_primary_2_mirrors(), 1, 1, + False, 10) + self.subject.logger = Mock(spec=['log', 'warn', 'info', 'debug', 'error', 'warning', 'fatal']) + + self.mock_logger = self.get_mock_from_apply_patch('logger') def tearDown(self): super(RebalanceSegmentsTestCase, self).tearDown() @@ -58,6 +72,50 @@ def test_rebalance_returns_failure(self): result = self.subject.rebalance() self.assertFalse(result) + # cherry-pick from Greenplum Database + # greenplum-db/gpdb-archive @ 482967c1b4, + # gpMgmt/bin/gppylib/test/unit/test_unit_rebalance_segment.py + @patch('gppylib.db.dbconn.querySingleton', return_value='56780000000') + def test_rebalance_returns_warning(self, mock1): + with self.assertRaises(Exception) as ex: + self.subject.rebalance() + self.assertEqual('56780000000 bytes of wal is still to be replayed on mirror with dbid 2, let mirror catchup ' + 'on replay then trigger rebalance. Use --replay-lag to configure the allowed replay lag limit ' + 'or --disable-replay-lag to disable the check completely if you wish to continue with ' + 'rebalance anyway', str(ex.exception)) + self.assertEqual([call("Get replay lag on mirror of primary segment with host:sdw1, port:40000")], + self.mock_logger.debug.call_args_list) + self.assertEqual([call("Determining primary and mirror segment pairs to rebalance"), + call('Allowed replay lag during rebalance is 10 GB')], + self.subject.logger.info.call_args_list) + + @patch('gppylib.db.dbconn.querySingleton', return_value='5678000000') + def test_rebalance_does_not_return_warning(self, mock1): + self.subject.rebalance() + self.assertEqual([call("Get replay lag on mirror of primary segment with host:sdw1, port:40000")], + self.mock_logger.debug.call_args_list) + + @patch('gppylib.db.dbconn.querySingleton', return_value='56780000000') + def test_rebalance_replay_lag_is_disabled(self, mock1): + self.subject.disable_replay_lag = True + self.subject.rebalance() + self.assertNotIn([call("Get replay lag on mirror of primary segment with host:sdw1, port:40000")], + self.mock_logger.debug.call_args_list) + self.assertIn([call("Determining primary and mirror segment pairs to rebalance")], + self.subject.logger.info.call_args_list) + + @patch('gppylib.db.dbconn.connect', side_effect=Exception()) + def test_replay_lag_connect_exception(self, mock1): + with self.assertRaises(Exception) as ex: + replay_lag(self.primary0) + self.assertEqual('Failed to query pg_stat_replication for host:sdw1, port:40000, error: ', str(ex.exception)) + + @patch('gppylib.db.dbconn.querySingleton', side_effect=Exception()) + def test_replay_lag_query_exception(self, mock1): + with self.assertRaises(Exception) as ex: + replay_lag(self.primary0) + self.assertEqual('Failed to query pg_stat_replication for host:sdw1, port:40000, error: ', str(ex.exception)) + def _create_gparray_with_2_primary_2_mirrors(self): coordinator = Segment.initFromString( "1|-1|p|p|s|u|cdw|cdw|5432|/data/coordinator") From ef894d4c421e699a7361b8db55aed961c101ae37 Mon Sep 17 00:00:00 2001 From: Dianjin Wang Date: Sat, 10 Oct 2026 11:26:34 +0800 Subject: [PATCH 6/7] gprecoverseg: port --max-rate from Greenplum A full recovery runs pg_basebackup, which will use whatever network bandwidth it can get. On a cluster that is still serving queries that can starve everything else, so Greenplum lets the operator cap it with --max-rate, validated and then passed through to pg_basebackup. Cloudberry has the same plumbing -- gprecoverseg builds a GpMirrorListToBuild, which builds a GpSegRecovery command line, which gpsegrecovery parses and turns into a PgBaseBackup -- but no maxRate parameter anywhere along it. Add it at each hop, along with the validation, the two "this flag is ignored with -r/-o" warnings, and the "Maximum Transfer Rate" row in the recovery plan the scenario asserts on. This lets the scenario run instead of being skipped: gprecoverseg recovers segment for valid max-rate options and errors out for others cherry-pick from Greenplum Database, greenplum-db/gpdb-archive @ 482967c1b4: gpMgmt/bin/gppylib/programs/clsRecoverSegment.py gpMgmt/bin/gppylib/operations/buildMirrorSegments.py gpMgmt/bin/gppylib/commands/gp.py gpMgmt/bin/gppylib/commands/pg.py gpMgmt/sbin/gpsegrecovery.py gpMgmt/sbin/recovery_base.py Two deliberate departures from upstream. Greenplum also threads maxRate into its differential recovery path, which Cloudberry does not have. And Cloudberry's FullRecovery retries pg_basebackup with --create-slot when the first attempt fails on a cluster that never had mirrors; upstream has no such retry, so the cap is applied to it too -- without that the retry would silently ignore the limit the operator asked for. The comment in gprecoverseg.feature that claimed Greenplum does not implement this either was wrong; it goes when #2170 drops the tag. Split out of #1639 (originally 993b99d5758). --- gpMgmt/bin/gppylib/commands/gp.py | 8 +- gpMgmt/bin/gppylib/commands/pg.py | 6 +- .../gppylib/operations/buildMirrorSegments.py | 10 ++- .../unit/test_unit_buildMirrorSegments.py | 4 +- .../bin/gppylib/programs/clsRecoverSegment.py | 76 ++++++++++++++++++- .../test/unit/test_unit_gprecoverseg.py | 2 + .../test/unit/test_unit_gpsegrecovery.py | 36 ++++----- .../test/unit/test_unit_recovery_base.py | 4 +- .../test/unit/test_unit_recoveryinfo.py | 4 +- gpMgmt/sbin/gpsegrecovery.py | 17 +++-- gpMgmt/sbin/recovery_base.py | 1 + 11 files changed, 132 insertions(+), 36 deletions(-) diff --git a/gpMgmt/bin/gppylib/commands/gp.py b/gpMgmt/bin/gppylib/commands/gp.py index d5c5646b61c..0a2944b48a0 100644 --- a/gpMgmt/bin/gppylib/commands/gp.py +++ b/gpMgmt/bin/gppylib/commands/gp.py @@ -1039,12 +1039,12 @@ class GpSegRecovery(Command): """ Format gpsegrecovery call for running pg_basebackup/pg_rewind on remoteHost for the segments passed in confinfo """ - def __init__(self, name, confinfo, logdir, batchSize, verbose, remoteHost, forceoverwrite, era): - cmdStr = _get_cmd_for_recovery_wrapper('gpsegrecovery', confinfo, logdir, batchSize, verbose, forceoverwrite, era) + def __init__(self, name, confinfo, logdir, batchSize, verbose, remoteHost, forceoverwrite, era, maxRate): + cmdStr = _get_cmd_for_recovery_wrapper('gpsegrecovery', confinfo, logdir, batchSize, verbose, forceoverwrite, era, maxRate) Command.__init__(self, name, cmdStr, REMOTE, remoteHost, start_new_session=True) -def _get_cmd_for_recovery_wrapper(wrapper_filename, confinfo, logdir, batchSize, verbose, forceoverwrite, era=None): +def _get_cmd_for_recovery_wrapper(wrapper_filename, confinfo, logdir, batchSize, verbose, forceoverwrite, era=None, maxRate=None): cmdStr = '$GPHOME/sbin/{}.py -c {} -l {}'.format(wrapper_filename, shlex.quote(confinfo), shlex.quote(logdir)) if verbose: @@ -1055,6 +1055,8 @@ def _get_cmd_for_recovery_wrapper(wrapper_filename, confinfo, logdir, batchSize, cmdStr += " --force-overwrite" if era: cmdStr += " --era={}".format(era) + if maxRate: + cmdStr += " --max-rate {}".format(maxRate) return cmdStr diff --git a/gpMgmt/bin/gppylib/commands/pg.py b/gpMgmt/bin/gppylib/commands/pg.py index 676f240e037..187b937a44b 100644 --- a/gpMgmt/bin/gppylib/commands/pg.py +++ b/gpMgmt/bin/gppylib/commands/pg.py @@ -208,7 +208,7 @@ def __init__(self, name, target_datadir, source_host, source_port, progress_file class PgBaseBackup(Command): def __init__(self, target_datadir, source_host, source_port, create_slot=False, replication_slot_name=None, excludePaths=[], ctxt=LOCAL, remoteHost=None, forceoverwrite=False, target_gp_dbid=0, - progress_file=None, recovery_mode=True): + progress_file=None, recovery_mode=True, max_rate=None): cmd_tokens = ['pg_basebackup', '-c', 'fast'] cmd_tokens.append('-D') cmd_tokens.append(target_datadir) @@ -253,6 +253,10 @@ def __init__(self, target_datadir, source_host, source_port, create_slot=False, cmd_tokens.append('-E') cmd_tokens.append(path) + if max_rate: + cmd_tokens.append('--max-rate') + cmd_tokens.append(max_rate) + cmd_tokens.append('--progress') cmd_tokens.append('--verbose') diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 1c5997833ae..f2418a90a17 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -135,7 +135,7 @@ class Action: ADDMIRRORS='add' RECOVERMIRRORS='recover' - def __init__(self, toBuild, pool, quiet, parallelDegree, additionalWarnings=None, logger=logger, forceoverwrite=False, progressMode=Progress.INPLACE, parallelPerHost=gp.DEFAULT_SEGHOST_NUM_WORKERS): + def __init__(self, toBuild, pool, quiet, parallelDegree, additionalWarnings=None, logger=logger, forceoverwrite=False, progressMode=Progress.INPLACE, parallelPerHost=gp.DEFAULT_SEGHOST_NUM_WORKERS, maxRate=None): self.__mirrorsToBuild = toBuild self.__pool = pool self.__quiet = quiet @@ -144,6 +144,7 @@ def __init__(self, toBuild, pool, quiet, parallelDegree, additionalWarnings=None # true for gprecoverseg and gpmovemirrors; false for gpexpand and gpaddmirrors self.__forceoverwrite = forceoverwrite self.__parallelPerHost = parallelPerHost + self.__maxRate = maxRate self.__additionalWarnings = additionalWarnings or [] self.segments_to_mark_down = [] if not logger: @@ -174,6 +175,12 @@ def getAdditionalWarnings(self): """ return self.__additionalWarnings + def getMaxTransferRate(self): + """ + returns the maximum transfer rate of data directory during Full recovery of failed segments + """ + return self.__maxRate + def _cleanup_before_recovery(self, gpArray, gpEnv): self.checkForPortAndDirectoryConflicts(gpArray) self._stop_failed_segments(gpEnv) @@ -528,6 +535,7 @@ def _do_recovery(self, recovery_info_by_host, gpEnv): batchSize=self.__parallelPerHost, remoteHost=hostName, era=era, + maxRate=self.__maxRate, forceoverwrite=self.__forceoverwrite)) completed_recovery_results = self.__runWaitAndCheckWorkerPoolForErrorsAndClear(cmds, suppressErrorCheck=True, progressCmds=progress_cmds) diff --git a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py index 0154b83d139..126d11a254d 100644 --- a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py @@ -105,9 +105,9 @@ def _setup_build_mirrors_mocks(self, build_mirrors_obj): def _setup_recovery_mocks(self, host_errors=[]): host1_recovery_output = gp.GpSegRecovery('test recover 1', 'dummy_info1', '/tmp/log1/', False, 1, - 'host1', 'dummy_era1', True) + 'host1', 'dummy_era1', True, None) host2_recovery_output = gp.GpSegRecovery('test recover 2', 'dummy_info2', '/tmp/log2/', True, 2, - 'host2', 'dummy_era2', False) + 'host2', 'dummy_era2', False, None) host1_result = CommandResult(0, ''.encode(), ''.encode(), True, False) host2_result = CommandResult(0, ''.encode(), ''.encode(), True, False) diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py index 890ecf855b5..d3a8b323744 100644 --- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py +++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py @@ -19,7 +19,7 @@ from gppylib.mainUtils import * from optparse import OptionGroup -import glob, os, sys, signal, shutil, time +import glob, os, re, sys, signal, shutil, time from contextlib import closing from gppylib import gparray, gplog, userinput, utils @@ -46,6 +46,13 @@ logger = gplog.get_default_logger() +# cherry-pick from Greenplum Database +# greenplum-db/gpdb-archive @ 482967c1b4, +# gpMgmt/bin/gppylib/programs/clsRecoverSegment.py +# Upper bound and lower bound for max-rate +MAX_RATE_LOWER = 32 +MAX_RATE_UPPER = 1048576 + # ------------------------------------------------------------------------- class RemoteQueryCommand(Command): @@ -133,7 +140,8 @@ def getRecoveryActionsBasedOnOptions(self, gpEnv, gpArray): instance.getInterfaceHostnameWarnings(), forceoverwrite=True, progressMode=self.getProgressMode(), - parallelPerHost=self.__options.parallelPerHost) + parallelPerHost=self.__options.parallelPerHost, + maxRate=self.__options.maxRate) def syncPackages(self, new_hosts): # The design decision here is to squash any exceptions resulting from the @@ -198,6 +206,10 @@ def displayRecovery(self, mirrorBuilder, gpArray): toRecover.getFailoverSegment(), tabLog) else: tabLog.info(["Recovery Target", "= in-place"]) + + if syncMode == "Full" and mirrorBuilder.getMaxTransferRate() is not None: + tabLog.info(["Maximum Transfer Rate", "= " + mirrorBuilder.getMaxTransferRate()]) + tabLog.outputTable() i = i + 1 @@ -231,6 +243,14 @@ def _getRecoveryWarnings(self, mirrorBuilder): "primary %s failover target: %s" % (self.__getSimpleSegmentLabel(src), self.__getSimpleSegmentLabel(dest))) + if not toRecover.isFullSynchronization() and mirrorBuilder.getMaxTransferRate() is not None: + # + # For Incremental recovery, when max-rate option is provided, + # warn the user that only Full recovery supports max-rate option + # + res.append(" --max-rate flag is only supported with segments undergoing Full recovery (-F). " + "Other modes of recovery will use the entire available network bandwidth.") + for warning in mirrorBuilder.getAdditionalWarnings(): res.append(warning) @@ -242,6 +262,46 @@ def _get_dblist(self): res = dbconn.query(conn, "SELECT datname FROM PG_DATABASE WHERE datname != 'template0'") return res.fetchall() + # cherry-pick from Greenplum Database + # greenplum-db/gpdb-archive @ 482967c1b4, + # gpMgmt/bin/gppylib/programs/clsRecoverSegment.py + def validateMaxRate(self): + """ + Validate the max-rate input provided by the user. + Numeric part of max-rate can be a whole/decimal number with an optional suffix of either 'k' or 'M' + Valid range is from 32kbps to 1048576kbps(1024Mbps) + + Below regex pattern validates the max-rate which can be a whole/decimal number with an optional unit suffix. + Valid match patterns - '12', '12.34A', '123.45' + Invalid match patterns - ABC', '.', '12 34' + + """ + # (\d+(?:\.\d+)?), not ([\d\.]+): the looser upstream form also matches + # '.', '...' and '1.2.3', which float() then rejects with a ValueError + # that escapes this function -- the operator gets "could not convert + # string to float" instead of the message below. The docstring above + # already calls '.' invalid; this makes the pattern agree with it. + pattern = r'^\s*(\d+(?:\.\d+)?)([a-zA-Z]?)\s*$' + match = re.match(pattern, self.__options.maxRate) + + if match: + rateVal, rateSuffix = match.groups() + + rateVal = float(rateVal) + if(rateVal <= 0): + raise ProgramArgumentValidationException("Transfer rate must be greater than zero") + + if rateSuffix == 'M': + rateVal *= 1024 + elif rateSuffix and rateSuffix != 'k': + raise ProgramArgumentValidationException("Invalid --max-rate unit: {0}".format(rateSuffix)) + + if rateVal < MAX_RATE_LOWER or rateVal > MAX_RATE_UPPER: + raise ProgramArgumentValidationException("transfer rate {0} is out of range".format(self.__options.maxRate)) + else: + raise ProgramArgumentValidationException("transfer rate {0} is not a valid value".format(self.__options.maxRate)) + return + def run(self): if self.__options.parallelDegree < 1 or self.__options.parallelDegree > gp.MAX_COORDINATOR_NUM_WORKERS: raise ProgramArgumentValidationException( @@ -286,6 +346,16 @@ def run(self): if self.__options.newRecoverHosts is not None: raise ProgramArgumentValidationException("-F option is not supported with -p option") + # Verify max-rate supported options + if self.__options.maxRate and self.__options.rebalanceSegments: + self.logger.warn(" -r flag does not support the use of --max-rate, hence --max-rate will be ignored") + if self.__options.outputSampleConfigFile and self.__options.maxRate: + self.logger.warn(" -o flag does not support the use of --max-rate, hence --max-rate will be ignored") + + # Validate max-rate value, if provided as an option + if self.__options.maxRate: + self.validateMaxRate() + faultProberInterface.getFaultProber().initializeProber(gpEnv.getCoordinatorPort()) confProvider = configInterface.getConfigurationProvider().initializeProvider(gpEnv.getCoordinatorPort()) @@ -535,6 +605,8 @@ def createParser(): dest='disableReplayLag', help='Disable replay lag check when rebalancing segments') addTo.add_option('', '--hba-hostnames', action='store_true', dest='hba_hostnames', help='use hostnames instead of CIDR in pg_hba.conf') + addTo.add_option('--max-rate', type='string', dest='maxRate', metavar='', + help='Maximum Rate of data transfer') parser.set_defaults() return parser diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py index cdf3f8a21ce..1575661eddd 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py @@ -34,6 +34,7 @@ def __init__(self): self.quiet = None self.interactive = False self.hba_hostnames = False + self.maxRate = None class GpRecoversegTestCase(GpTestCase): @@ -90,6 +91,7 @@ def setUp(self): patch.object(GpMirrorListToBuild, "recover_mirrors"), patch.object(GpMirrorListToBuild, "getAdditionalWarnings"), patch.object(GpMirrorListToBuild, "getMirrorsToBuild"), + patch.object(GpMirrorListToBuild, "getMaxTransferRate"), patch.object(HeapChecksum, "check_segment_consistency"), patch.object(HeapChecksum, "get_segments_checksum_settings"), ]) diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py index bd380107c7d..11c8617398d 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py @@ -133,9 +133,10 @@ def setUp(self): p.getSegmentPort(), True, '/tmp/test_progress_file') self.era = '1234_20211110' + self.maxRate = '1024M' self.full_recovery_cmd = gpsegrecovery.FullRecovery( name='test full recovery', recovery_info=self.seg_recovery_info, - forceoverwrite=True, logger=self.mock_logger, era=self.era) + forceoverwrite=True, logger=self.mock_logger, era=self.era, maxRate=self.maxRate) def tearDown(self): super(FullRecoveryTestCase, self).tearDown() @@ -172,7 +173,7 @@ def test_basebackup_run_passes(self): expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self._assert_basebackup_runs(expected_init_args1) self._assert_cmd_passed() @@ -184,7 +185,7 @@ def test_basebackup_run_no_forceoverwrite_passes(self): expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=False, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=False, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self._assert_basebackup_runs(expected_init_args1) self._assert_cmd_passed() @@ -195,10 +196,10 @@ def test_basebackup_run_one_exception(self): expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') expected_init_args2 = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self.assertEqual(2, self.mock_pgbasebackup_init.call_count) self.assertEqual([expected_init_args1, expected_init_args2] , self.mock_pgbasebackup_init.call_args_list) self.assertEqual(2, self.mock_pgbasebackup_run.call_count) @@ -214,10 +215,10 @@ def test_basebackup_run_two_exceptions(self): expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') expected_init_args2 = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self.assertEqual(2, self.mock_pgbasebackup_init.call_count) self.assertEqual([expected_init_args1, expected_init_args2], self.mock_pgbasebackup_init.call_args_list) self.assertEqual(2, self.mock_pgbasebackup_run.call_count) @@ -237,11 +238,11 @@ def test_basebackup_run_no_forceoverwrite_two_exceptions(self): expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=False, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=False, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') # regardless of the passed in value, second call to pg_basebackup will always have forceoverwrite=True expected_init_args2 = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self.assertEqual(2, self.mock_pgbasebackup_init.call_count) self.assertEqual([expected_init_args1, expected_init_args2], self.mock_pgbasebackup_init.call_args_list) self.assertEqual(2, self.mock_pgbasebackup_run.call_count) @@ -256,7 +257,7 @@ def test_basebackup_init_exception(self): self.full_recovery_cmd.run() expected_init_args = call("/data/mirror0", "sdw1", '40000', create_slot=False, replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file') + forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self.assertEqual(1, self.mock_pgbasebackup_init.call_count) self.assertEqual(expected_init_args, self.mock_pgbasebackup_init.call_args) self.assertEqual(0, self.mock_pgbasebackup_run.call_count) @@ -289,6 +290,7 @@ def setUp(self): self.incr_r2 = RecoveryInfo('target_data_dir4', 5004, 4, 'source_hostname4', 6004, False, '/tmp/progress_file4') self.era = '1234_2021110' + self.maxRate = '1024M' self.apply_patches([ patch('gpsegrecovery.SegmentStart.__init__', return_value=None), @@ -355,10 +357,10 @@ def test_complete_workflow_exception(self, mock_pgbasebackup_run, mock_pgbasebac def test_get_recovery_cmds_is_called(self, mock_get_recovery_cmds, mock_recovery_base_main, mock_logger): mix_confinfo = gppylib.recoveryinfo.serialize_list([self.full_r1, self.incr_r2]) sys.argv = ['gpsegrecovery', '-l', '/tmp/logdir', '--era={}'.format(self.era), '-f', - '-c {}'.format(mix_confinfo)] + '-c {}'.format(mix_confinfo), '--max-rate={}'.format(self.maxRate)] SegRecovery().main() mock_get_recovery_cmds.assert_called_once_with([self.full_r1, self.incr_r2], True, mock_logger.return_value, - self.era) + self.era, self.maxRate) mock_recovery_base_main.assert_called_once_with(mock_get_recovery_cmds.return_value) def _assert_validation_full_call(self, cmd, expected_recovery_info, @@ -380,26 +382,26 @@ def _assert_setup_incr_call(self, cmd, expected_recovery_info): self.assertEqual(self.mock_logger, cmd.logger) def test_empty_recovery_info_list(self): - cmd_list = SegRecovery().get_recovery_cmds([], False, None, self.era) + cmd_list = SegRecovery().get_recovery_cmds([], False, None, self.era, self.maxRate) self.assertEqual([], cmd_list) def test_get_recovery_cmds_full_recoveryinfo(self): - cmd_list = SegRecovery().get_recovery_cmds([self.full_r1, self.full_r2], False, self.mock_logger, self.era) + cmd_list = SegRecovery().get_recovery_cmds([self.full_r1, self.full_r2], False, self.mock_logger, self.era, self.maxRate) self._assert_validation_full_call(cmd_list[0], self.full_r1) self._assert_validation_full_call(cmd_list[1], self.full_r2) def test_get_recovery_cmds_incr_recoveryinfo(self): - cmd_list = SegRecovery().get_recovery_cmds([self.incr_r1, self.incr_r2], False, self.mock_logger, self.era) + cmd_list = SegRecovery().get_recovery_cmds([self.incr_r1, self.incr_r2], False, self.mock_logger, self.era, self.maxRate) self._assert_setup_incr_call(cmd_list[0], self.incr_r1) self._assert_setup_incr_call(cmd_list[1], self.incr_r2) def test_get_recovery_cmds_mix_recoveryinfo(self): - cmd_list = SegRecovery().get_recovery_cmds([self.full_r1, self.incr_r2], False, self.mock_logger, self.era) + cmd_list = SegRecovery().get_recovery_cmds([self.full_r1, self.incr_r2], False, self.mock_logger, self.era, self.maxRate) self._assert_validation_full_call(cmd_list[0], self.full_r1) self._assert_setup_incr_call(cmd_list[1], self.incr_r2) def test_get_recovery_cmds_mix_recoveryinfo_forceoverwrite(self): - cmd_list = SegRecovery().get_recovery_cmds([self.full_r1, self.incr_r2], True, self.mock_logger, self.era) + cmd_list = SegRecovery().get_recovery_cmds([self.full_r1, self.incr_r2], True, self.mock_logger, self.era, self.maxRate) self._assert_validation_full_call(cmd_list[0], self.full_r1, expected_forceoverwrite=True) self._assert_setup_incr_call(cmd_list[1], self.incr_r2) diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_recovery_base.py b/gpMgmt/bin/gppylib/test/unit/test_unit_recovery_base.py index 8d7f7c7a316..f5af9b6a439 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_recovery_base.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_recovery_base.py @@ -324,7 +324,7 @@ def test_decorator(cmd): raise Exception('running the cmd failed') recovery_info = RecoveryInfo('/tmp/datadir2', 7002, 2, None, None, None, '/tmp/progress_file2') - test_cmd = FullRecovery('original name', recovery_info, True,None,None) + test_cmd = FullRecovery('original name', recovery_info, True,None,None,None) test_decorator(test_cmd) self.assertEqual('new name', test_cmd.name) @@ -339,7 +339,7 @@ def test_decorator(cmd): cmd.error_type = None raise Exception('running the cmd failed') recovery_info = RecoveryInfo('/tmp/datadir2', 7002, 2, None, None, None, '/tmp/progress_file2') - test_cmd = FullRecovery('original name', recovery_info, True,None,None) + test_cmd = FullRecovery('original name', recovery_info, True,None,None,None) test_decorator(test_cmd) self.assertEqual('new name', test_cmd.name) diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_recoveryinfo.py b/gpMgmt/bin/gppylib/test/unit/test_unit_recoveryinfo.py index b744500a67b..d497ada68df 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_recoveryinfo.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_recoveryinfo.py @@ -408,8 +408,8 @@ def run_tests(self, tests, run_recovery=False): else: host2_result = CommandResult(1, 'failed 2'.encode(), test["host2_error"].encode(), True, False) - host1_recovery_output = gp.GpSegRecovery(None, None, None, False, 1, 'host1', None, True) - host2_recovery_output = gp.GpSegRecovery(None, None, None, False, 1, 'host2', None, True) + host1_recovery_output = gp.GpSegRecovery(None, None, None, False, 1, 'host1', None, True, None) + host2_recovery_output = gp.GpSegRecovery(None, None, None, False, 1, 'host2', None, True, None) host1_recovery_output.get_results = Mock(return_value=host1_result) host2_recovery_output.get_results = Mock(return_value=host2_result) diff --git a/gpMgmt/sbin/gpsegrecovery.py b/gpMgmt/sbin/gpsegrecovery.py index a1a6891ee35..f602a784f74 100644 --- a/gpMgmt/sbin/gpsegrecovery.py +++ b/gpMgmt/sbin/gpsegrecovery.py @@ -13,12 +13,13 @@ class FullRecovery(Command): - def __init__(self, name, recovery_info, forceoverwrite, logger, era): + def __init__(self, name, recovery_info, forceoverwrite, logger, era, maxRate): self.name = name self.recovery_info = recovery_info self.replicationSlotName = 'internal_wal_replication_slot' 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) @@ -37,7 +38,8 @@ def run(self): replication_slot_name=self.replicationSlotName, forceoverwrite=self.forceoverwrite, target_gp_dbid=self.recovery_info.target_segment_dbid, - progress_file=self.recovery_info.progress_file) + 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) try: cmd.run(validateAfter=True) @@ -56,7 +58,8 @@ def run(self): replication_slot_name=self.replicationSlotName, forceoverwrite=True, target_gp_dbid=self.recovery_info.target_segment_dbid, - progress_file=self.recovery_info.progress_file) + progress_file=self.recovery_info.progress_file, + max_rate=self.maxRate) self.logger.info("Re-running pg_basebackup, creating the slot this time") cmd.run(validateAfter=True) @@ -145,9 +148,10 @@ def signal_handler(sig, frame): signal.signal(signal.SIGTERM, signal_handler) recovery_base.main(self.get_recovery_cmds(recovery_base.seg_recovery_info_list, recovery_base.options.forceoverwrite, - recovery_base.logger, recovery_base.options.era)) + recovery_base.logger, recovery_base.options.era, + recovery_base.options.maxRate)) - def get_recovery_cmds(self, seg_recovery_info_list, forceoverwrite, logger, era): + def get_recovery_cmds(self, seg_recovery_info_list, forceoverwrite, logger, era, maxRate): cmd_list = [] for seg_recovery_info in seg_recovery_info_list: if seg_recovery_info.is_full_recovery: @@ -155,7 +159,8 @@ def get_recovery_cmds(self, seg_recovery_info_list, forceoverwrite, logger, era) recovery_info=seg_recovery_info, forceoverwrite=forceoverwrite, logger=logger, - era=era) + era=era, + maxRate=maxRate) else: cmd = IncrementalRecovery(name='Run pg_rewind', recovery_info=seg_recovery_info, diff --git a/gpMgmt/sbin/recovery_base.py b/gpMgmt/sbin/recovery_base.py index 36bbb0d13ba..0a6e76b34e8 100644 --- a/gpMgmt/sbin/recovery_base.py +++ b/gpMgmt/sbin/recovery_base.py @@ -39,6 +39,7 @@ def parseargs(self): parser.add_option('-f', '--force-overwrite', dest='forceoverwrite', action='store_true', default=False) parser.add_option('-l', '--log-dir', dest="logfileDirectory", type="string") parser.add_option('', '--era', dest="era", help="coordinator era", ) + parser.add_option('--max-rate', type='string', dest='maxRate', metavar='') # Parse the command line arguments self.options, _ = parser.parse_args() From b85b91bd4141c88a95d11d2415f9499974663e8d Mon Sep 17 00:00:00 2001 From: Dianjin Wang Date: Sat, 10 Oct 2026 12:04:23 +0800 Subject: [PATCH 7/7] gprecoverseg: drop a stale replication slot instead of failing into a retry A full recovery streams from the primary through the replication slot internal_wal_replication_slot. pg_basebackup --create-slot fails if that slot already exists, and PostgreSQL offers no way to reuse one. Cloudberry worked around it by guessing: run pg_basebackup without --create-slot, and if anything at all goes wrong, run the whole thing again with it. That is expensive when the backup failed for an unrelated reason -- the second attempt also forces forceoverwrite=True, discarding the target directory -- and it leaves the slot handling implicit. The code says as much: GPDB_12_MERGE_FIXME could we check it before? or let pg_basebackup create slot if not exists. Greenplum answers that FIXME: ask the primary whether the slot is there, drop it if so, and pass --create-slot once. Port PgReplicationSlot and use it, in gpsegrecovery for gprecoverseg and in gpconfigurenewsegment for gpexpand, dropping both retries. Porting it into gpconfigurenewsegment is not optional. PgBaseBackup's constructor now talks to the database, and there the second constructor call sat inside an except block, so a failure to reach the primary would have escaped from the handler rather than being reported. This lets the last two skipped scenarios run, which also assert the messages this prints: gprecoverseg should drop existing slot on full recovery recovery should not try to drop slot if slot does not exist With them, nothing in gprecoverseg.feature or gpmovemirrors.feature is tagged @not_implemented any more. cherry-pick from Greenplum Database, greenplum-db/gpdb-archive @ 482967c1b4: gpMgmt/bin/gppylib/commands/pg.py gpMgmt/sbin/gpsegrecovery.py gpMgmt/bin/lib/gpconfigurenewsegment Two things in gpconfigurenewsegment are fixed in passing, both found by running open-code-review over this commit: - It called PgBaseBackup with logfile=, a keyword that constructor has never had, so the mirror path raised TypeError before doing anything. Upstream has the same line and the same bug; nothing in either test suite reaches it, because gpaddmirrors builds mirrors through gpsegrecovery and only gpexpand comes this way. It is progress_file=. - The constructor call sat outside the try below. Now that it talks to the database it can fail, and a failure there returned no result at all, which the worker pool reports as "'NoneType' object has no attribute 'wasSuccessful'" long after the fact. One deliberate departure from upstream: drop_slot() catches pgdb's DatabaseError, not psycopg2's, because that is the driver Cloudberry's dbconn is built on. Split out of #1639 (originally 83778564c0f). --- gpMgmt/bin/gppylib/commands/pg.py | 73 +++++++++++++++++- .../test/unit/test_unit_pg_base_backup.py | 9 ++- .../test/unit/test_unit_gpsegrecovery.py | 74 ++++++------------- gpMgmt/bin/lib/gpconfigurenewsegment | 50 +++++-------- gpMgmt/sbin/gpsegrecovery.py | 24 +----- 5 files changed, 121 insertions(+), 109 deletions(-) diff --git a/gpMgmt/bin/gppylib/commands/pg.py b/gpMgmt/bin/gppylib/commands/pg.py index 187b937a44b..50dc1312dae 100644 --- a/gpMgmt/bin/gppylib/commands/pg.py +++ b/gpMgmt/bin/gppylib/commands/pg.py @@ -8,6 +8,9 @@ from gppylib.gplog import * from gppylib.gparray import * +from gppylib.db import dbconn +from contextlib import closing +from pgdb import DatabaseError from .base import * from .unix import * from gppylib.commands.base import * @@ -205,6 +208,63 @@ def __init__(self, name, target_datadir, source_host, source_port, progress_file Command.__init__(self, name, self.cmdStr, LOCAL) +# cherry-pick from Greenplum Database +# greenplum-db/gpdb-archive @ 482967c1b4, gpMgmt/bin/gppylib/commands/pg.py +class PgReplicationSlot: + """ + PgReplicationSlot have utility function related to replication slot + """ + def __init__(self, host, port, slot_name): + """ + @param host + @param port + @param slot_name + """ + self.host = host + self.port = port + self.name = slot_name + + def slot_exists(self): + count = -1 + logger.debug('Checking if slot {} exists for host:{}, port:{}'.format(self.name, self.host, self.port)) + sql = "SELECT count(*) FROM pg_catalog.pg_replication_slots WHERE slot_name = '{}'". format(self.name) + try: + dburl = dbconn.DbURL(hostname=self.host, port=self.port) + with closing(dbconn.connect(dburl, utility=True, encoding='UTF8')) as conn: + count = dbconn.querySingleton(conn, sql) + except Exception as ex: + raise Exception("Failed to query pg_replication_slots for host:{}, port:{}: {}". + format(self.host, self.port, str(ex))) + + if count == 0: + logger.debug("Slot {} does not exist for host:{}, port:{}". + format(self.name, self.host, self.port)) + return False + + return True + + def drop_slot(self): + logger.debug("Dropping slot {} for host:{}, port:{}".format(self.name, self.host, self.port)) + sql = "SELECT pg_drop_replication_slot('{}');".format(self.name) + try: + dburl = dbconn.DbURL(hostname=self.host, port=self.port) + with closing(dbconn.connect(dburl, utility=True, encoding='UTF8')) as conn: + dbconn.query(conn, sql) + except DatabaseError as e: + # one of the case can be where slot is present but currently in active state + logger.exception("Failed to query pg_drop_replication_slot for host:{}, port:{}: {}". + format(self.host, self.port, str(e))) + return False + except Exception as ex: + raise Exception("Failed to drop replication slot for host:{}, port:{} : {}". + format(self.host, self.port, str(ex))) + + logger.debug("Successfully dropped replication slot {} for host:{}, port:{}". + format(self.name, self.host, self.port)) + + return True + + class PgBaseBackup(Command): def __init__(self, target_datadir, source_host, source_port, create_slot=False, replication_slot_name=None, excludePaths=[], ctxt=LOCAL, remoteHost=None, forceoverwrite=False, target_gp_dbid=0, @@ -217,8 +277,19 @@ def __init__(self, target_datadir, source_host, source_port, create_slot=False, cmd_tokens.append('-p') cmd_tokens.append(source_port) + # if there is already slot present and create-slot arg is true it will give error, + # there is no option available in upstream postgres so that existing slot can be reused + # it's good choice to drop a existing available slot and recreate the new one + # if the slot does not already exist or drop slot is success we will create a new slot + # but if we are not able to drop the slot in that case, + # we will consider it as an error and will avoid creating a new slot if create_slot: - cmd_tokens.append('--create-slot') + pg_slot = PgReplicationSlot(source_host, source_port, replication_slot_name) + if pg_slot.slot_exists(): + if pg_slot.drop_slot(): + cmd_tokens.append('--create-slot') + else: + cmd_tokens.append('--create-slot') cmd_tokens.extend(self._xlog_arguments(replication_slot_name)) diff --git a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_pg_base_backup.py b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_pg_base_backup.py index 117f62b41ea..2f45a87a547 100644 --- a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_pg_base_backup.py +++ b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_pg_base_backup.py @@ -1,4 +1,5 @@ import unittest +from mock import patch from gppylib.commands import pg @@ -19,7 +20,9 @@ def test_replication_slot_not_passed_when_not_given_slot_name(self): self.assertIn("fetch", tokens) self.assertNotIn("stream", tokens) - def test_base_backup_passes_parameters_necessary_to_create_replication_slot_when_given_slotname(self): + @patch('gppylib.commands.pg.PgReplicationSlot.slot_exists', return_value=True) + @patch('gppylib.commands.pg.PgReplicationSlot.drop_slot', return_value=True) + def test_base_backup_passes_parameters_necessary_to_create_replication_slot_when_given_slotname(self, mock1, mock2): base_backup = pg.PgBaseBackup( create_slot=True, replication_slot_name='some-replication-slot-name', @@ -32,8 +35,10 @@ def test_base_backup_passes_parameters_necessary_to_create_replication_slot_when self.assertIn("some-replication-slot-name", base_backup.command_tokens) self.assertIn("--wal-method", base_backup.command_tokens) self.assertIn("stream", base_backup.command_tokens) + self.assertIn("--create-slot", base_backup.command_tokens) - def test_base_backup_does_not_pass_conflicting_xlog_method_argument_when_given_replication_slot(self): + @patch('gppylib.commands.pg.PgReplicationSlot.slot_exists', return_value=False) + def test_base_backup_does_not_pass_conflicting_xlog_method_argument_when_given_replication_slot(self, mock1): base_backup = pg.PgBaseBackup( create_slot=True, replication_slot_name='some-replication-slot-name', diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py index 11c8617398d..abca6b3745f 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py @@ -171,7 +171,7 @@ def _assert_cmd_failed(self, expected_stderr): def test_basebackup_run_passes(self): self.full_recovery_cmd.run() - expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, + expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') @@ -183,7 +183,7 @@ def test_basebackup_run_no_forceoverwrite_passes(self): self.full_recovery_cmd.run() - expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, + expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', forceoverwrite=False, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self._assert_basebackup_runs(expected_init_args1) @@ -194,68 +194,39 @@ def test_basebackup_run_one_exception(self): self.full_recovery_cmd.run() - expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, + expected_init_args = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') - expected_init_args2 = call("/data/mirror0", "sdw1", '40000', create_slot=True, - replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') - self.assertEqual(2, self.mock_pgbasebackup_init.call_count) - self.assertEqual([expected_init_args1, expected_init_args2] , self.mock_pgbasebackup_init.call_args_list) - self.assertEqual(2, self.mock_pgbasebackup_run.call_count) - self.assertEqual([call(validateAfter=True),call(validateAfter=True)], self.mock_pgbasebackup_run.call_args_list) - gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era) - self._assert_cmd_passed() - - def test_basebackup_run_two_exceptions(self): - self.mock_pgbasebackup_run.side_effect=[Exception('backup failed once'), - Exception('backup failed twice')] - - self.full_recovery_cmd.run() - - expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, - replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') - expected_init_args2 = call("/data/mirror0", "sdw1", '40000', create_slot=True, - replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') - self.assertEqual(2, self.mock_pgbasebackup_init.call_count) - self.assertEqual([expected_init_args1, expected_init_args2], self.mock_pgbasebackup_init.call_args_list) - self.assertEqual(2, self.mock_pgbasebackup_run.call_count) - self.assertEqual([call(validateAfter=True),call(validateAfter=True)], self.mock_pgbasebackup_run.call_args_list) - self.mock_logger.info.any_call('Running pg_basebackup failed: backup failed once') - self.mock_logger.info.assert_called_with("Re-running pg_basebackup, creating the slot this time") + self.assertEqual(1, self.mock_pgbasebackup_init.call_count) + self.assertEqual([expected_init_args], self.mock_pgbasebackup_init.call_args_list) + self.assertEqual(1, self.mock_pgbasebackup_run.call_count) + self.assertEqual([call(validateAfter=True)], self.mock_pgbasebackup_run.call_args_list) self.assertEqual(0, gpsegrecovery.start_segment.call_count) - self._assert_cmd_failed('{"error_type": "full", "error_msg": "backup failed twice", "dbid": 2, ' \ + self._assert_cmd_failed('{"error_type": "full", "error_msg": "backup failed once", "dbid": 2, ' \ '"datadir": "/data/mirror0", "port": 50000, "progress_file": "/tmp/test_progress_file"}') - def test_basebackup_run_no_forceoverwrite_two_exceptions(self): - self.mock_pgbasebackup_run.side_effect = [Exception('backup failed once'), - Exception('backup failed twice')] - self.full_recovery_cmd.forceoverwrite = False + def test_basebackup_run_no_forceoverwrite_one_exceptions(self): + self.mock_pgbasebackup_run.side_effect = [Exception('backup failed once'), Mock()] + self.full_recovery_cmd.forceoverwrite = False self.full_recovery_cmd.run() - expected_init_args1 = call("/data/mirror0", "sdw1", '40000', create_slot=False, + expected_init_args = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', forceoverwrite=False, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') - # regardless of the passed in value, second call to pg_basebackup will always have forceoverwrite=True - expected_init_args2 = call("/data/mirror0", "sdw1", '40000', create_slot=True, - replication_slot_name='internal_wal_replication_slot', - forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') - self.assertEqual(2, self.mock_pgbasebackup_init.call_count) - self.assertEqual([expected_init_args1, expected_init_args2], self.mock_pgbasebackup_init.call_args_list) - self.assertEqual(2, self.mock_pgbasebackup_run.call_count) - self.assertEqual([call(validateAfter=True),call(validateAfter=True)], self.mock_pgbasebackup_run.call_args_list) + self.assertEqual(1, self.mock_pgbasebackup_init.call_count) + self.assertEqual([expected_init_args], self.mock_pgbasebackup_init.call_args_list) + self.assertEqual(1, self.mock_pgbasebackup_run.call_count) + self.assertEqual([call(validateAfter=True)], self.mock_pgbasebackup_run.call_args_list) self.assertEqual(0, gpsegrecovery.start_segment.call_count) - self._assert_cmd_failed('{"error_type": "full", "error_msg": "backup failed twice", "dbid": 2, ' \ + self._assert_cmd_failed('{"error_type": "full", "error_msg": "backup failed once", "dbid": 2, ' \ '"datadir": "/data/mirror0", "port": 50000, "progress_file": "/tmp/test_progress_file"}') def test_basebackup_init_exception(self): self.mock_pgbasebackup_init.side_effect = [Exception('backup init failed')] self.full_recovery_cmd.run() - expected_init_args = call("/data/mirror0", "sdw1", '40000', create_slot=False, + expected_init_args = call("/data/mirror0", "sdw1", '40000', create_slot=True, replication_slot_name='internal_wal_replication_slot', forceoverwrite=True, target_gp_dbid=2, progress_file='/tmp/test_progress_file', max_rate='1024M') self.assertEqual(1, self.mock_pgbasebackup_init.call_count) @@ -329,8 +300,7 @@ def test_complete_workflow(self, mock_pgbasebackup_run, mock_pgbasebackup_init, def test_complete_workflow_exception(self, mock_pgbasebackup_run, mock_pgbasebackup_init, mock_pgrewind_run, mock_pgrewind_init): mock_pgrewind_run.side_effect = [Exception('pg_rewind failed')] - mock_pgbasebackup_run.side_effect = [Exception('pg_basebackup failed once'), - Exception('pg_basebackup failed twice')] + mock_pgbasebackup_run.side_effect = [Exception('pg_basebackup failed')] mix_confinfo = gppylib.recoveryinfo.serialize_list([ self.full_r1, self.incr_r2]) sys.argv = ['gpsegrecovery', '-l', '/tmp/logdir', '--era={}'.format(self.era), '-c {}'.format(mix_confinfo)] @@ -341,14 +311,14 @@ def test_complete_workflow_exception(self, mock_pgbasebackup_run, mock_pgbasebac self.assertCountEqual('[{"error_type": "incremental", "error_msg": "pg_rewind failed", "dbid": 4, "datadir": "target_data_dir4", ' '"port": 5004, "progress_file": "/tmp/progress_file4"} , ' - '{"error_type": "full", "error_msg": "pg_basebackup failed twice", "dbid": 1,' + '{"error_type": "full", "error_msg": "pg_basebackup failed", "dbid": 1,' '"datadir": "target_data_dir1", "port": 5001, "progress_file": "/tmp/progress_file1"}]', buf.getvalue().strip()) self.assertEqual(1, ex.exception.code) self.assertEqual(1, mock_pgrewind_run.call_count) self.assertEqual(1, mock_pgrewind_init.call_count) - self.assertEqual(2, mock_pgbasebackup_run.call_count) - self.assertEqual(2, mock_pgbasebackup_init.call_count) + self.assertEqual(1, mock_pgbasebackup_run.call_count) + self.assertEqual(1, mock_pgbasebackup_init.call_count) self.assertRegex(gplog.get_logfile(), r'/gpsegrecovery.py_\d+\.log') @patch('recovery_base.gplog.setup_tool_logging') diff --git a/gpMgmt/bin/lib/gpconfigurenewsegment b/gpMgmt/bin/lib/gpconfigurenewsegment index c37c70bf1fa..26a6c4d96b7 100755 --- a/gpMgmt/bin/lib/gpconfigurenewsegment +++ b/gpMgmt/bin/lib/gpconfigurenewsegment @@ -134,45 +134,31 @@ class ConfExpSegCmd(Command): self.progressFile = '%s/pg_basebackup.%s.dbid%s.out' % (gplog.get_logger_dir(), datetime.datetime.today().strftime('%Y%m%d_%H%M%S'), self.dbid) - # Create a mirror based on the primary - cmd = PgBaseBackup(target_datadir=self.datadir, - source_host=self.syncWithSegmentHostname, - source_port=str(self.syncWithSegmentPort), - create_slot = False, - replication_slot_name=self.replicationSlotName, - forceoverwrite=self.forceoverwrite, - target_gp_dbid=self.dbid, - logfile=self.progressFile) + # Create a mirror based on the primary. + # + # Built inside the try: PgBaseBackup's constructor now asks + # the source segment whether the replication slot exists, so + # it can fail, and a failure out here would leave the command + # with no result at all -- the pool then reports + # "'NoneType' object has no attribute 'wasSuccessful'" + # instead of saying what went wrong. try: + cmd = PgBaseBackup(target_datadir=self.datadir, + source_host=self.syncWithSegmentHostname, + source_port=str(self.syncWithSegmentPort), + create_slot=True, + replication_slot_name=self.replicationSlotName, + forceoverwrite=self.forceoverwrite, + target_gp_dbid=self.dbid, + progress_file=self.progressFile) logger.info("Running pg_basebackup with progress output temporarily in %s" % self.progressFile) cmd.run(validateAfter=True) self.set_results(CommandResult(0, b'', b'', True, False)) if shouldDeleteProgressFile: os.remove(self.progressFile) - except Exception as e: - # If the cluster never has mirrors, cmd will fail - # quickly because the internal slot doesn't exist. - # Re-run with `create_slot`. - # GPDB_12_MERGE_FIXME could we check it before? or let - # pg_basebackup create slot if not exists. - cmd = PgBaseBackup(target_datadir=self.datadir, - source_host=self.syncWithSegmentHostname, - source_port=str(self.syncWithSegmentPort), - create_slot = True, - replication_slot_name=self.replicationSlotName, - forceoverwrite=True, - target_gp_dbid=self.dbid, - logfile=self.progressFile) - try: - logger.info("Re-running pg_basebackup, creating the slot this time") - cmd.run(validateAfter=True) - self.set_results(CommandResult(0, b'', b'', True, False)) - if shouldDeleteProgressFile: - os.remove(self.progressFile) - except Exception as e: - self.set_results(CommandResult(1, b'', str(e).encode(), True, False)) - raise + self.set_results(CommandResult(1, b'', str(e).encode(), True, False)) + raise logger.info("Successfully ran pg_basebackup: %s" % cmd.cmdStr) return diff --git a/gpMgmt/sbin/gpsegrecovery.py b/gpMgmt/sbin/gpsegrecovery.py index f602a784f74..5b0834eedac 100644 --- a/gpMgmt/sbin/gpsegrecovery.py +++ b/gpMgmt/sbin/gpsegrecovery.py @@ -34,34 +34,14 @@ def run(self): cmd = PgBaseBackup(self.recovery_info.target_datadir, self.recovery_info.source_hostname, str(self.recovery_info.source_port), - create_slot=False, + 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) - try: - cmd.run(validateAfter=True) - except Exception as e: #TODO should this be ExecutionError? - self.logger.info("Running pg_basebackup failed: {}".format(str(e))) - - # If the cluster never has mirrors, cmd will fail - # quickly because the internal slot doesn't exist. - # Re-run with `create_slot`. - # GPDB_12_MERGE_FIXME could we check it before? or let - # pg_basebackup create slot if not exists. - 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=True, - target_gp_dbid=self.recovery_info.target_segment_dbid, - progress_file=self.recovery_info.progress_file, - max_rate=self.maxRate) - self.logger.info("Re-running pg_basebackup, creating the slot this time") - cmd.run(validateAfter=True) + cmd.run(validateAfter=True) self.error_type = RecoveryErrorType.DEFAULT_ERROR self.logger.info("Successfully ran pg_basebackup for dbid: {}".format(