diff --git a/gpMgmt/bin/gppylib/commands/gp.py b/gpMgmt/bin/gppylib/commands/gp.py index b980bc629c4..0a2944b48a0 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. @@ -1034,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: @@ -1050,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 @@ -1670,6 +1677,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/pg.py b/gpMgmt/bin/gppylib/commands/pg.py index 676f240e037..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,10 +208,67 @@ 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, - 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) @@ -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)) @@ -253,6 +324,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/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/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/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 869a37d312b..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) @@ -240,6 +247,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 +303,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 +461,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 +504,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): @@ -518,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/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/operations/test/unit/test_unit_buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py index 61a88c21539..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) @@ -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..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): @@ -122,7 +129,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()] @@ -131,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 @@ -196,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 @@ -229,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) @@ -240,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( @@ -262,6 +324,38 @@ 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") + + 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. + 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") + + # 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()) @@ -378,12 +472,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("********************************") @@ -500,8 +598,15 @@ 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') + 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/programs/clsRecoverSegment_triples.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py index 34b6a12ff2e..674387f9807 100644 --- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py +++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py @@ -1,11 +1,99 @@ 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.utils import checkNotNone, normalizeAndValidateInputPath, validateHostnameAddress 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))) + + +# 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: @@ -84,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 @@ -142,12 +232,36 @@ 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 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(!). @@ -155,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 @@ -166,10 +289,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 @@ -222,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) @@ -285,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']): @@ -292,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) @@ -322,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/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: 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_gprecoverseg.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gprecoverseg.py index 292a6b8b5f1..1575661eddd 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 @@ -32,6 +34,7 @@ def __init__(self): self.quiet = None self.interactive = False self.hba_hostnames = False + self.maxRate = None class GpRecoversegTestCase(GpTestCase): @@ -88,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 983b5f25a9a..abca6b3745f 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") @@ -122,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() @@ -135,8 +147,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): @@ -154,9 +171,9 @@ 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') + 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() @@ -166,9 +183,9 @@ 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') + 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() @@ -177,70 +194,41 @@ 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') - 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') - 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') - 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') - 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") + 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_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, - replication_slot_name='internal_wal_replication_slot', - forceoverwrite=False, target_gp_dbid=2, progress_file='/tmp/test_progress_file') - # 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, + 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') - 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) + forceoverwrite=False, 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_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') + 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) @@ -273,10 +261,12 @@ 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), patch('gpsegrecovery.SegmentStart.run'), + patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()), ]) self.mock_segment_start_init = self.get_mock_from_apply_patch('__init__') @@ -310,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)] @@ -322,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') @@ -338,10 +327,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, @@ -363,26 +352,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_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") 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/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 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 811a73ccbb5..5b0834eedac 100644 --- a/gpMgmt/sbin/gpsegrecovery.py +++ b/gpMgmt/sbin/gpsegrecovery.py @@ -7,18 +7,19 @@ 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 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) @@ -33,36 +34,22 @@ 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) + 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) - 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( 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 +74,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) @@ -124,9 +128,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: @@ -134,7 +139,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()