gprecoverseg: fix the defects the Behave suite exposes

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):
    """

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()

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):

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()
 

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("********************************")