Make it possible to apply some recovery params without restart (#1260)

Starting from PostgreSQL 12 the following recovery parameters could be changed without restart, but Patroni didn't yet support it:
* archive_cleanup_command
* promote_trigger_file
* recovery_end_command
* recovery_min_apply_delay

In future postgres releases this list will be extended and Patroni will support it automatically.

patroni/ha.py

@@ -380,12 +380,17 @@ class Ha(object):
         # In this case it is safe to continue running without changing recovery.conf
         if self.is_standby_cluster() and role == 'replica' and not (node_to_follow and node_to_follow.conn_url):
             return 'continue following the old known standby leader'
-        elif not self.state_handler.config.check_recovery_conf(node_to_follow):
-            self._async_executor.try_run_async('changing primary_conninfo and restarting',
-                                               self.state_handler.follow, args=(node_to_follow, role))
-        elif role == 'standby_leader' and self.state_handler.role != role:
-            self.state_handler.set_role(role)
-            self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE)
+        else:
+            change_required, restart_required = self.state_handler.config.check_recovery_conf(node_to_follow)
+            if change_required:
+                if restart_required:
+                    self._async_executor.try_run_async('changing primary_conninfo and restarting',
+                                                       self.state_handler.follow, args=(node_to_follow, role))
+                else:
+                    self.state_handler.follow(node_to_follow, role, do_reload=True)
+            elif role == 'standby_leader' and self.state_handler.role != role:
+                self.state_handler.set_role(role)
+                self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE)
 
         return follow_reason
 

patroni/postgresql/__init__.py

@@ -684,7 +684,7 @@ class Postgresql(object):
         except Exception:
             logger.exception('Failed to read and parse %s', (history_path,))
 
-    def follow(self, member, role='replica', timeout=None):
+    def follow(self, member, role='replica', timeout=None, do_reload=False):
         recovery_params = self.config.build_recovery_params(member)
         self.config.write_recovery_conf(recovery_params)
 
@@ -698,7 +698,11 @@ class Postgresql(object):
             self.__cb_pending = ACTION_NOOP
 
         if self.is_running():
-            self.restart(block_callbacks=change_role, role=role)
+            if do_reload:
+                self.config.write_postgresql_conf()
+                self.reload()
+            else:
+                self.restart(block_callbacks=change_role, role=role)
         else:
             self.start(timeout=timeout, block_callbacks=change_role, role=role)
 

patroni/postgresql/config.py

@@ -468,6 +468,7 @@ class ConfigHandler(object):
         keywords = ('user', 'passfile', 'host', 'port', 'sslmode', 'sslcompression', 'sslcert',
                     'sslkey', 'sslrootcert', 'sslcrl', 'application_name', 'krbsrvname')
         if include_dbname:
+            params = params.copy()
             params['dbname'] = params.get('database') or self._postgresql.database
             keywords = ('dbname',) + keywords
 
@@ -541,7 +542,8 @@ class ConfigHandler(object):
             return None, False
 
         try:
-            values = {p[0]: p[1] for p in self._get_pg_settings(self._recovery_parameters_to_compare).values()}
+            values = self._get_pg_settings(self._recovery_parameters_to_compare).values()
+            values = {p[0]: [p[1], p[4] == 'postmaster'] for p in values}
             self._postgresql_conf_mtime = pg_conf_mtime
             self._auto_conf_mtime = auto_conf_mtime
             self._postmaster_ctime = postmaster_ctime
@@ -567,10 +569,10 @@ class ConfigHandler(object):
                     value = read_recovery_param_value(line[match.end():])
                 if value is None:
                     return None, True
-                values[match.group(1)] = value
+                values[match.group(1)] = [value, True]
             self._recovery_conf_mtime = recovery_conf_mtime
-        values.setdefault('recovery_min_apply_delay', '0')
-        values.update({param: '' for param in self._recovery_parameters_to_compare if param not in values})
+        values.setdefault('recovery_min_apply_delay', ['0', True])
+        values.update({param: ['', True] for param in self._recovery_parameters_to_compare if param not in values})
         return values, True
 
     def _check_passfile(self, passfile, wanted_primary_conninfo):
@@ -616,12 +618,12 @@ class ConfigHandler(object):
         # TODO: recovery.conf could be stale, would be nice to detect that.
         if self._postgresql.major_version >= 120000:
             if not os.path.exists(self._standby_signal):
-                return False
+                return True, True
 
             _read_recovery_params = self._read_recovery_params
         else:
             if not self.recovery_conf_exists():
-                return False
+                return True, True
 
             _read_recovery_params = self._read_recovery_params_pre_v12
 
@@ -630,31 +632,36 @@ class ConfigHandler(object):
         # was changed and params were read either from the config or from the database connection.
         if updated:
             if params is None:  # exception or unparsable config
-                return False
+                return True, True
 
             # We will cache parsed value until the next config change.
             self._current_recovery_params = params
-            if params['primary_conninfo']:
-                params['primary_conninfo'] = parse_dsn(params['primary_conninfo'])
+            primary_conninfo = params['primary_conninfo']
+            if primary_conninfo[0]:
+                primary_conninfo[0] = parse_dsn(params['primary_conninfo'][0])
                 # If we failed to parse non-empty connection string this indicates that config if broken.
-                if not params['primary_conninfo']:
-                    return False
+                if not primary_conninfo[0]:
+                    return True, True
             else:  # empty string, primary_conninfo is not in the config
-                params['primary_conninfo'] = {}
+                primary_conninfo[0] = {}
+
+        required = {'restart': 0, 'reload': 0}
+
+        def record_missmatch(mtype):
+            required['restart' if mtype else 'reload'] += 1
 
-        ret = True
         wanted_recovery_params = self.build_recovery_params(member)
         for param, value in self._current_recovery_params.items():
             if param == 'recovery_min_apply_delay':
-                if not compare_values('integer', 'ms', value, wanted_recovery_params.get(param, 0)):
-                    ret = False
+                if not compare_values('integer', 'ms', value[0], wanted_recovery_params.get(param, 0)):
+                    record_missmatch(value[1])
             elif param == 'primary_conninfo':
-                if not self._check_primary_conninfo(value, wanted_recovery_params.get('primary_conninfo', {})):
-                    ret = False
+                if not self._check_primary_conninfo(value[0], wanted_recovery_params.get('primary_conninfo', {})):
+                    record_missmatch(value[1])
             elif (param != 'primary_slot_name' or wanted_recovery_params.get('primary_conninfo')) \
-                    and str(value) != str(wanted_recovery_params.get(param, '')):
-                ret = False
-        return ret
+                    and str(value[0]) != str(wanted_recovery_params.get(param, '')):
+                record_missmatch(value[1])
+        return required['restart'] + required['reload'] > 0, required['restart'] > 0
 
     @staticmethod
     def _remove_file_if_exists(name):

tests/test_ha.py

@@ -159,6 +159,7 @@ def run_async(self, func, args=()):
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_pgpass', Mock(return_value={}))
 @patch.object(ConfigHandler, 'write_recovery_conf', Mock())
+@patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
 @patch.object(Postgresql, 'query', Mock())
 @patch.object(Postgresql, 'checkpoint', Mock())
 @patch.object(CancellableSubprocess, 'call', Mock(return_value=0))
@@ -358,6 +359,7 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
         self.ha.patroni.replicatefrom = "foo"
+        self.p.config.check_recovery_conf = Mock(return_value=(True, False))
         self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
 
     def test_follow_in_pause(self):
@@ -683,7 +685,7 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.p.name = 'leader'
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
-        self.p.config.check_recovery_conf = true
+        self.p.config.check_recovery_conf = Mock(return_value=(False, False))
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
         self.assertEqual(self.ha.run_cycle(), 'no action.  i am the standby leader with the lock')
 

tests/test_postgresql.py

@@ -208,24 +208,24 @@ class TestPostgresql(BaseTestPostgresql):
             'primary_conninfo': ['primary_conninfo', 'foo=', None, 'string', 'postmaster'],
             'recovery_min_apply_delay': ['recovery_min_apply_delay', '0', 'ms', 'integer', 'sighup']
         }
-        self.assertFalse(self.p.config.check_recovery_conf(None))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
         self.p.config.write_recovery_conf({'standby_mode': 'on'})
-        self.assertFalse(self.p.config.check_recovery_conf(None))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
         mock_get_pg_settings.return_value['primary_conninfo'][1] = ''
         mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '1'
-        self.assertFalse(self.p.config.check_recovery_conf(None))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (True, False))
         mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '0'
-        self.assertTrue(self.p.config.check_recovery_conf(None))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))
         conninfo = {'host': '1', 'password': 'bar'}
         with patch('patroni.postgresql.config.ConfigHandler.primary_conninfo_params', Mock(return_value=conninfo)):
             mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '1'
-            self.assertFalse(self.p.config.check_recovery_conf(None))
+            self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
             mock_get_pg_settings.return_value['primary_conninfo'][1] = 'host=1 passfile=' + self.p.config._pgpass
             mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '0'
-            self.assertFalse(self.p.config.check_recovery_conf(None))
+            self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
             self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': conninfo.copy()})
             self.p.config.write_postgresql_conf()
-            self.assertTrue(self.p.config.check_recovery_conf(None))
+            self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))
 
     @patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000))
     @patch.object(Postgresql, 'is_running', MockPostmaster)
@@ -235,20 +235,20 @@ class TestPostgresql(BaseTestPostgresql):
         mock_get_pg_settings.return_value = {'primary_conninfo': ['primary_conninfo', '', None, 'string', 'postmaster']}
         self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
         self.p.config.write_postgresql_conf()
-        self.assertTrue(self.p.config.check_recovery_conf(None))
-        self.assertTrue(self.p.config.check_recovery_conf(None))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))
         mock_get_pg_settings.side_effect = Exception
         with patch('patroni.postgresql.config.mtime', mock_mtime):
-            self.assertFalse(self.p.config.check_recovery_conf(None))
+            self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
 
     @patch.object(Postgresql, 'major_version', PropertyMock(return_value=100000))
     def test__read_recovery_params_pre_v12(self):
         self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
-        self.assertFalse(self.p.config.check_recovery_conf(None))
-        self.assertFalse(self.p.config.check_recovery_conf(None))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
+        self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
         self.p.config.write_recovery_conf({'standby_mode': '\n'})
         with patch('patroni.postgresql.config.mtime', mock_mtime):
-            self.assertFalse(self.p.config.check_recovery_conf(None))
+            self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
 
     def test_write_postgresql_and_sanitize_auto_conf(self):
         read_data = 'primary_conninfo = foo\nfoo = bar\n'