diff --git a/gpMgmt/bin/gppylib/commands/gp.py b/gpMgmt/bin/gppylib/commands/gp.py index b980bc629c4..03f6c1269e7 100644 --- a/gpMgmt/bin/gppylib/commands/gp.py +++ b/gpMgmt/bin/gppylib/commands/gp.py @@ -1670,6 +1670,18 @@ def createTempDirectoryName(coordinatorDataDirectory, tempDirPrefix): datetime.datetime.now().strftime('%m%d%Y'), os.getpid()) +# Cherry-picked from Greenplum Database (greenplum-db/gpdb-archive @ 482967c1b4), +# gpMgmt/bin/gppylib/commands/gp.py. +def is_gprecoverseg_running(): + gprecoverseg_pidfile = os.path.join(get_coordinatordatadir(), 'gprecoverseg.lock', 'PID') + try: + with open(gprecoverseg_pidfile) as pidfile: + gprecoverseg_pid = pidfile.read() + except Exception: + return False + + return check_pid(gprecoverseg_pid) + #------------------------------------------------------------------------- class GpRecoverSeg(Command): """ diff --git a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py index 4ac75c69685..018641a6682 100644 --- a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py +++ b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py @@ -4,9 +4,9 @@ # from gppylib.commands.base import CommandResult -from mock import patch +from mock import patch, mock_open -from gppylib.commands.gp import is_pid_postmaster, get_postmaster_pid_locally +from gppylib.commands.gp import is_pid_postmaster, get_postmaster_pid_locally, is_gprecoverseg_running from test.unit.gp_unittest import GpTestCase, run_tests @@ -158,5 +158,22 @@ def test_get_postmaster_pid_locally_error(self, mock1): def test_get_postmaster_pid_locally_empty(self, mock1): self.assertEqual(get_postmaster_pid_locally('/tmp'), -1) + @patch('gppylib.commands.gp.check_pid', return_value=True) + @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/data/coordinator') + @patch("builtins.open", new_callable=mock_open, read_data="123") + def test_is_gprecoverseg_running_succeeds(self, mock_file, mock1, mock2): + result = is_gprecoverseg_running() + mock_file.assert_called_once_with('/data/coordinator/gprecoverseg.lock/PID') + mock2.assert_called_once_with('123') + self.assertTrue(result) + + @patch('gppylib.commands.gp.check_pid') + @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/invalid/path/') + def test_is_gprecoverseg_running_when_pidfile_does_not_exists(self, mock1, mock2): + result = is_gprecoverseg_running() + self.assertFalse(result) + self.assertFalse(mock2.called) + + if __name__ == '__main__': run_tests() diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py index 869a37d312b..1c5997833ae 100644 --- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py @@ -240,6 +240,9 @@ def __build_mirrors(self, actionName, gpEnv, gpArray): recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild) + # Remove any existing progress files for segments to be recovered + self.remove_existing_progress_files(recovery_info_by_host) + self._run_setup_recovery(actionName, recovery_info_by_host) backout_map = self._update_config(recovery_info_by_host, gpArray) @@ -293,13 +296,17 @@ def _update_config(self, recovery_info_by_host, gpArray): return backout_map - def _remove_progress_files(self, recovery_info_by_host, recovery_results): + # Remove any existing progress file of segments that will be recovered by + # current gprecoverseg execution. The files written by this run are left in + # place: gpstate reads them to report progress, and they are what an + # operator looks at after a recovery that went wrong. + def remove_existing_progress_files(self, recovery_info_by_host): remove_progress_file_cmds = [] for hostName, recovery_info_list in recovery_info_by_host.items(): for ri in recovery_info_list: - if recovery_results.was_bb_rewind_successful(ri.target_segment_dbid): - remove_progress_file_cmds.append(self._get_remove_cmd(ri.progress_file, hostName)) - self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=False) + remove_progress_file_cmds.append(self._get_remove_cmd("*dbid{}.out".format(ri.target_segment_dbid), + hostName)) + self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=True) def _revert_config_update(self, recovery_results, backout_map): if len(backout_map) == 0: @@ -447,8 +454,12 @@ def _get_progress_cmd(self, progressFile, targetSegmentDbId, targetHostname): remoteHost=targetHostname) return None + # remove_file is a name pattern (e.g. "*dbid2.out"), matched in the log + # directory the progress files are written to. def _get_remove_cmd(self, remove_file, target_host): - return base.Command("remove file", "rm -f {}".format(shlex.quote(remove_file)), ctxt=base.REMOTE, remoteHost=target_host) + return base.Command("remove file", "find {} -name {} -delete".format(shlex.quote(gplog.get_logger_dir()), + shlex.quote(remove_file)), + ctxt=base.REMOTE, remoteHost=target_host) def __runWaitAndCheckWorkerPoolForErrorsAndClear(self, cmds, suppressErrorCheck=False, progressCmds=[]): for cmd in cmds: @@ -486,7 +497,6 @@ def _run_recovery(self, action_name, recovery_info_by_host, gpEnv): recovery_results = RecoveryResult(action_name, completed_recovery_results, self.__logger) recovery_results.print_bb_rewind_and_start_errors() - self._remove_progress_files(recovery_info_by_host, recovery_results) return recovery_results def _do_recovery(self, recovery_info_by_host, gpEnv): diff --git a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py index 61a88c21539..0154b83d139 100644 --- a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py +++ b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py @@ -522,6 +522,24 @@ def test_run_setup_recovery_empty_errors(self): self._assert_setup_recovery() + def test_remove_existing_progress_files(self): + self._setup_recovery_mocks() + + recovery_info = OrderedDict() + recovery_info['host1'] = [self.recovery_info1] + recovery_info['host2'] = [self.recovery_info2, self.recovery_info3] + + self.default_build_mirrors_obj.remove_existing_progress_files(recovery_info) + + self.assertEqual([call(ANY, suppressErrorCheck=True)], + self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) + rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][0][0] + self.assertEqual(['host1', 'host2', 'host2'], [cmd.remoteHost for cmd in rm_progress_cmds]) + self.assertEqual(["find /tmp/logdir -name '*dbid2.out' -delete", + "find /tmp/logdir -name '*dbid3.out' -delete", + "find /tmp/logdir -name '*dbid4.out' -delete"], + [cmd.cmdStr for cmd in rm_progress_cmds]) + def test_run_recovery_no_errors(self): self._setup_recovery_mocks() @@ -531,12 +549,11 @@ def test_run_recovery_no_errors(self): self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) seg_recovery_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][0][0] progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][1]['progressCmds'] - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] #TODO fix formatting expected_recovery_cmd_strs = ["""$GPHOME/sbin/gpsegrecovery.py -c '[{"target_datadir": "/datadir2", "target_port": 7001, "target_segment_dbid": 2, "source_hostname": "source_host_for_dbid2", "source_port": 7002, "is_full_recovery": true, "progress_file": "/tmp/progress_file2"}]' -l /tmp/logdir -b 64 --era=dummy_era""", """$GPHOME/sbin/gpsegrecovery.py -c '[{"target_datadir": "/datadir3", "target_port": 7003, "target_segment_dbid": 3, "source_hostname": "source_host_for_dbid3", "source_port": 7004, "is_full_recovery": true, "progress_file": "/tmp/progress_file3"}, {"target_datadir": "/datadir4", "target_port": 7005, "target_segment_dbid": 4, "source_hostname": "source_host_for_dbid4", "source_port": 7006, "is_full_recovery": true, "progress_file": "/tmp/progress_file4"}]' -l /tmp/logdir -b 64 --era=dummy_era"""] @@ -558,13 +575,6 @@ def test_run_recovery_no_errors(self): progress_cmds[2].cmdStr) self.assertEqual(1, self.mock_gp_era.call_count) - self.assertEqual(3, len(rm_progress_cmds)) - self.assertEqual("host1", rm_progress_cmds[0].remoteHost) - self.assertEqual("rm -f /tmp/progress_file2", rm_progress_cmds[0].cmdStr) - self.assertEqual("host2", rm_progress_cmds[1].remoteHost) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[1].cmdStr) - self.assertEqual("host2", rm_progress_cmds[2].remoteHost) - self.assertEqual("rm -f /tmp/progress_file4", rm_progress_cmds[2].cmdStr) self._assert_run_recovery() @@ -584,11 +594,9 @@ def test_run_recovery_all_dbids_fail_only_bb_rewind_errors(self): self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(0, len(rm_progress_cmds)) self._assert_run_recovery() @@ -599,13 +607,9 @@ def test_run_recovery_some_dbids_fail_all_bb_rewind_errors(self): self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(1, len(rm_progress_cmds)) - self.assertEqual("host2", rm_progress_cmds[0].remoteHost) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[0].cmdStr) self._assert_run_recovery() @@ -616,11 +620,8 @@ def test_run_recovery_all_dbids_fail_all_start_errors(self): host2_error = '[{"error_type": "start", "error_msg":"some error for dbid 4", "dbid": 4, "datadir": "/datadir4", "port": 7005, "progress_file": "/tmp/progress4"}]' self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - - self.assertEqual(3, len(rm_cmds)) self._assert_run_recovery() def test_run_recovery_some_dbids_fail_all_start_errors(self): @@ -631,10 +632,8 @@ def test_run_recovery_some_dbids_fail_all_start_errors(self): self._setup_recovery_mocks([host1_error, host2_error]) self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(3, len(rm_progress_cmds)) self._assert_run_recovery() @@ -647,14 +646,10 @@ def test_run_recovery_all_dbids_fail_bb_rewind_and_start_errors(self): self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] # Since start passed for dbid:3, we should have deleted it's progress file - self.assertEqual(1, len(rm_progress_cmds)) - self.assertEqual("host1", rm_progress_cmds[0].remoteHost) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[0].cmdStr) self._assert_run_recovery() @@ -671,15 +666,9 @@ def test_run_recovery_all_dbids_fail_bb_rewind_and_default_errors(self): self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv) - self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)], + self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)], self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list) - rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0] - self.assertEqual(2, len(rm_progress_cmds)) - self.assertEqual("host1", rm_progress_cmds[0].remoteHost) - self.assertEqual("host2", rm_progress_cmds[1].remoteHost) - self.assertEqual("rm -f /tmp/progress_file2", rm_progress_cmds[0].cmdStr) - self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[1].cmdStr) self._assert_run_recovery() diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py index 81377136f83..503360368a4 100644 --- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py +++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py @@ -262,6 +262,25 @@ def run(self): if optionCnt > 1: raise ProgramArgumentValidationException("Only one of -i, -p, and -r may be specified") + # -o only writes out a sample configuration file describing what would + # be recovered; it does not recover anything, so combining it with an + # option that says where to recover to is a contradiction rather than a + # refinement. + if self.__options.outputSampleConfigFile: + if self.__options.recoveryConfigFile is not None: + raise ProgramArgumentValidationException("Invalid -i provided with -o argument") + if self.__options.rebalanceSegments: + raise ProgramArgumentValidationException("Invalid -r provided with -o argument") + + # A full resynchronisation rebuilds the mirror from its primary, which + # is neither what -r (rebalance back to preferred roles) nor -p + # (recover onto a different host) asks for. + if self.__options.forceFullResynchronization: + if self.__options.rebalanceSegments: + raise ProgramArgumentValidationException("-F option is not supported with -r option") + if self.__options.newRecoverHosts is not None: + raise ProgramArgumentValidationException("-F option is not supported with -p option") + faultProberInterface.getFaultProber().initializeProber(gpEnv.getCoordinatorPort()) confProvider = configInterface.getConfigurationProvider().initializeProvider(gpEnv.getCoordinatorPort()) @@ -378,12 +397,16 @@ def signal_handler(sig, frame): if not mirrorBuilder.recover_mirrors(gpEnv, gpArray): if self.termination_requested: self.logger.error("gprecoverseg process was interrupted by the user.") + self.logger.error("gprecoverseg failed. Please check the output for more details.") sys.exit(1) if self.termination_requested: self.logger.info("Not able to terminate the recovery process since it has been completed successfully.") self.logger.info("********************************") + self.logger.info("Future gprecoverseg executions might remove the currently created pg_basebackup/pg_rewind/rsync " + "progress files, please save these files if needed.") + self.logger.info("********************************") self.logger.info("Segments successfully recovered.") self.logger.info("********************************") diff --git a/gpMgmt/bin/gppylib/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_gpsegrecovery.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py index 983b5f25a9a..bd380107c7d 100644 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py @@ -18,11 +18,13 @@ def setUp(self): self.mock_logger = Mock() self.apply_patches([ patch('gpsegrecovery.start_segment', return_value=Mock()), + patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()), patch('gppylib.commands.pg.PgRewind.__init__', return_value=None), patch('gppylib.commands.pg.PgRewind.run') ]) self.mock_pgrewind_run = self.get_mock_from_apply_patch('run') self.mock_pgrewind_init = self.get_mock_from_apply_patch('__init__') + self.mock_modify_conf = self.get_mock_from_apply_patch('ModifyConfSetting') p = Segment.initFromString("1|0|p|p|s|u|sdw1|sdw1|40000|/data/primary0") m = Segment.initFromString("2|0|m|m|s|u|sdw2|sdw2|50000|/data/mirror0") @@ -56,8 +58,13 @@ def test_incremental_run_passes(self): self.assertEqual(1, self.mock_pgrewind_run.call_count) self.assertEqual(call(validateAfter=True), self.mock_pgrewind_run.call_args) logger_call_args = [call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file'), - call('Successfully ran pg_rewind for dbid: 2')] + call('Successfully ran pg_rewind for dbid: 2'), + call('Updating /data/mirror0/postgresql.conf')] self.assertEqual(logger_call_args, self.mock_logger.info.call_args_list) + self.mock_modify_conf.assert_called_once_with('Updating /data/mirror0/postgresql.conf', + '/data/mirror0/postgresql.conf', 'port', 50000, + optType='number') + self.mock_modify_conf.return_value.run.assert_called_once_with(validateAfter=True) gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era) def test_incremental_run_exception(self): @@ -71,6 +78,7 @@ def test_incremental_run_exception(self): self.assertEqual(call(validateAfter=True), self.mock_pgrewind_run.call_args) self.assertEqual([call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file')], self.mock_logger.info.call_args_list) + self.assertEqual(0, self.mock_modify_conf.call_count) self.assertEqual(0, gpsegrecovery.start_segment.call_count) self._assert_cmd_failed('{"error_type": "incremental", "error_msg": "pg_rewind failed", "dbid": 2, ' \ '"datadir": "/data/mirror0", "port": 50000, "progress_file": "/tmp/test_progress_file"}') @@ -92,7 +100,8 @@ def test_incremental_start_segment_exception(self): self.assertEqual(1, self.mock_pgrewind_init.call_count) self.assertEqual(1, self.mock_pgrewind_run.call_count) logger_call_args = [call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file'), - call('Successfully ran pg_rewind for dbid: 2')] + call('Successfully ran pg_rewind for dbid: 2'), + call('Updating /data/mirror0/postgresql.conf')] self.assertEqual(logger_call_args, self.mock_logger.info.call_args_list) gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era) self._assert_cmd_failed('{"error_type": "start", "error_msg": "pg_ctl start failed", "dbid": 2, ' \ @@ -107,11 +116,13 @@ def setUp(self): self.mock_logger = Mock(spec=['log', 'info', 'debug', 'error', 'warn', 'exception']) self.apply_patches([ patch('gpsegrecovery.start_segment', return_value=Mock()), + patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()), patch('gpsegrecovery.PgBaseBackup.__init__', return_value=None), patch('gpsegrecovery.PgBaseBackup.run') ]) self.mock_pgbasebackup_run = self.get_mock_from_apply_patch('run') self.mock_pgbasebackup_init = self.get_mock_from_apply_patch('__init__') + self.mock_modify_conf = self.get_mock_from_apply_patch('ModifyConfSetting') p = Segment.initFromString("1|0|p|p|s|u|sdw1|sdw1|40000|/data/primary0") m = Segment.initFromString("2|0|m|m|s|u|sdw2|sdw2|50000|/data/mirror0") @@ -135,8 +146,13 @@ def _assert_basebackup_runs(self, expected_init_args): self.assertEqual(1, self.mock_pgbasebackup_run.call_count) self.assertEqual(call(validateAfter=True), self.mock_pgbasebackup_run.call_args) expected_logger_info_args = [call('Running pg_basebackup with progress output temporarily in /tmp/test_progress_file'), - call("Successfully ran pg_basebackup for dbid: 2")] + call("Successfully ran pg_basebackup for dbid: 2"), + call('Updating /data/mirror0/postgresql.conf')] self.assertEqual(expected_logger_info_args, self.mock_logger.info.call_args_list) + self.mock_modify_conf.assert_called_once_with('Updating /data/mirror0/postgresql.conf', + '/data/mirror0/postgresql.conf', 'port', 50000, + optType='number') + self.mock_modify_conf.return_value.run.assert_called_once_with(validateAfter=True) gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era) def _assert_cmd_passed(self): @@ -277,6 +293,7 @@ def setUp(self): self.apply_patches([ patch('gpsegrecovery.SegmentStart.__init__', return_value=None), patch('gpsegrecovery.SegmentStart.run'), + patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()), ]) self.mock_segment_start_init = self.get_mock_from_apply_patch('__init__') diff --git a/gpMgmt/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/sbin/gpsegrecovery.py b/gpMgmt/sbin/gpsegrecovery.py index 811a73ccbb5..a1a6891ee35 100644 --- a/gpMgmt/sbin/gpsegrecovery.py +++ b/gpMgmt/sbin/gpsegrecovery.py @@ -7,7 +7,7 @@ from gppylib.commands.pg import PgBaseBackup, PgRewind from recovery_base import RecoveryBase, set_recovery_cmd_results from gppylib.commands.base import Command -from gppylib.commands.gp import SegmentStart +from gppylib.commands.gp import SegmentStart, ModifyConfSetting from gppylib.gparray import Segment from gppylib.commands.unix import terminate_proc_tree @@ -63,6 +63,10 @@ def run(self): self.error_type = RecoveryErrorType.DEFAULT_ERROR self.logger.info("Successfully ran pg_basebackup for dbid: {}".format( self.recovery_info.target_segment_dbid)) + + # Updating port number on conf after recovery + update_port_in_conf(self.recovery_info, self.logger) + self.error_type = RecoveryErrorType.START_ERROR start_segment(self.recovery_info, self.logger, self.era) @@ -87,10 +91,27 @@ def run(self): cmd.run(validateAfter=True) self.logger.info("Successfully ran pg_rewind for dbid: {}".format(self.recovery_info.target_segment_dbid)) + # Updating port number on conf after recovery + update_port_in_conf(self.recovery_info, self.logger) + self.error_type = RecoveryErrorType.START_ERROR start_segment(self.recovery_info, self.logger, self.era) +def update_port_in_conf(recovery_info, logger): + """ + pg_basebackup and pg_rewind both copy the source segment's + postgresql.conf, which carries the *source* port. Point it back at this + segment before it is started, or the mirror comes up on its primary's + port. + """ + logger.info("Updating %s/postgresql.conf" % recovery_info.target_datadir) + modifyConfCmd = ModifyConfSetting('Updating %s/postgresql.conf' % recovery_info.target_datadir, + "{}/{}".format(recovery_info.target_datadir, 'postgresql.conf'), + 'port', recovery_info.target_port, optType='number') + modifyConfCmd.run(validateAfter=True) + + def start_segment(recovery_info, logger, era): seg = Segment(None, None, None, None, None, None, None, None, recovery_info.target_port, recovery_info.target_datadir)