gprecoverseg: port the replay-lag check, --max-rate and stale-slot handling from Greenplum

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

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

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/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',

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)