Repository: zalando/patroni
Pull request #2975 (2023-12-06 08:41:34)

Reload postgres config if a server param was reset
Fix the case when a parameter value was changed and then reset back to the initial value without restart - before this fix, the second change was not reflected in the Postgres config.
This commit also includes the related unit test refactoring.


Patch:
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index 27cab778..fe66c6d8 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -666,7 +666,7 @@ class Consul(AbstractDCS):
         if ret:  # We have no other choise, only read after write :(
             if not retry.ensure_deadline(0.5):
                 return False
-            _, ret = self.retry(self._client.kv.get, self.sync_path)
+            _, ret = self.retry(self._client.kv.get, self.sync_path, consistency='consistent')
             if ret and (ret.get('Value') or b'').decode('utf-8') == value:
                 return ret['ModifyIndex']
         return False
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 271bbdfe..7dfa8934 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -1098,6 +1098,12 @@ class ConfigHandler(object):
                                     local_connection_address_changed = True
                             else:
                                 logger.info('Changed %s from %s to %s', r[0], r[1], new_value)
+                        elif r[0] in self._server_parameters \
+                                and not compare_values(r[3], r[2], r[1], self._server_parameters[r[0]]):
+                            # Check if any parameter was set back to the current pg_settings value
+                            # We can use pg_settings value here, as it is proved to be equal to new_value
+                            logger.info('Changed %s from %s to %s', r[0], self._server_parameters[r[0]], r[1])
+                            conf_changed = True
                 for param, value in changes.items():
                     if '.' in param:
                         # Check that user-defined-paramters have changed (parameters with period in name)
diff --git a/patroni/utils.py b/patroni/utils.py
index 23f419e5..2b1ded5b 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -400,22 +400,23 @@ def parse_real(value: Any, base_unit: Optional[str] = None) -> Optional[float]:
         return convert_to_base_unit(val, unit, base_unit)
 
 
-def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value: Any) -> bool:
-    """Check if *old_value* and *new_value* are equivalent after parsing them as *vartype*.
+def compare_values(vartype: str, unit: Optional[str], settings_value: Any, config_value: Any) -> bool:
+    """Check if the value from ``pg_settings`` and from Patroni config are equivalent after parsing them as *vartype*.
 
-    :param vartpe: the target type to parse *old_value* and *new_value* before comparing them. Accepts any among of the
-        following (case sensitive):
+    :param vartype: the target type to parse *settings_value* and *config_value* before comparing them.
+        Accepts any among of the following (case sensitive):
 
         * ``bool``: parse values using :func:`parse_bool`; or
         * ``integer``: parse values using :func:`parse_int`; or
         * ``real``: parse values using :func:`parse_real`; or
         * ``enum``: parse values as lowercase strings; or
         * ``string``: parse values as strings. This one is used by default if no valid value is passed as *vartype*.
-    :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real` for *new_value*.
-    :param old_value: value to be compared with *new_value*.
-    :param new_value: value to be compared with *old_value*.
+    :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real`
+        for *config_value*.
+    :param settings_value: value to be compared with *config_value*.
+    :param config_value: value to be compared with *settings_value*.
 
-    :returns: ``True`` if *old_value* is equivalent to *new_value* when both are parsed as *vartype*.
+    :returns: ``True`` if *settings_value* is equivalent to *config_value* when both are parsed as *vartype*.
 
     :Example:
 
@@ -455,8 +456,8 @@ def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value:
     }
 
     converter = converters.get(vartype) or converters['string']
-    old_converted = converter(old_value, None)
-    new_converted = converter(new_value, unit)
+    old_converted = converter(settings_value, None)
+    new_converted = converter(config_value, unit)
 
     return old_converted is not None and new_converted is not None and old_converted == new_converted
 


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index bd70ba3d..b013e4e1 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -25,8 +25,41 @@ mock_available_gucs = PropertyMock(return_value={
     'max_wal_senders', 'max_worker_processes', 'port', 'search_path', 'shared_preload_libraries',
     'stats_temp_directory', 'synchronous_standby_names', 'track_commit_timestamp', 'unix_socket_directories',
     'vacuum_cost_delay', 'vacuum_cost_limit', 'wal_keep_size', 'wal_level', 'wal_log_hints', 'zero_damaged_pages',
+    'autovacuum', 'wal_segment_size', 'wal_block_size', 'shared_buffers', 'wal_buffers',
 })
 
+GET_PG_SETTINGS_RESULT = [
+    ('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
+    ('wal_block_size', '8192', None, 'integer', 'internal'),
+    ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
+    ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
+    ('max_connections', '100', None, 'integer', 'postmaster'),
+    ('max_prepared_transactions', '200', None, 'integer', 'postmaster'),
+    ('max_worker_processes', '8', None, 'integer', 'postmaster'),
+    ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'),
+    ('max_wal_senders', '5', None, 'integer', 'postmaster'),
+    ('search_path', 'public', None, 'string', 'user'),
+    ('port', '5432', None, 'integer', 'postmaster'),
+    ('listen_addresses', '127.0.0.2, 127.0.0.3', None, 'string', 'postmaster'),
+    ('autovacuum', 'on', None, 'bool', 'sighup'),
+    ('unix_socket_directories', '/tmp', None, 'string', 'postmaster'),
+    ('shared_preload_libraries', 'citus', None, 'string', 'postmaster'),
+    ('wal_keep_size', '128', 'MB', 'integer', 'sighup'),
+    ('cluster_name', 'batman', None, 'string', 'postmaster'),
+    ('vacuum_cost_delay', '200', 'ms', 'real', 'user'),
+    ('vacuum_cost_limit', '-1', None, 'integer', 'user'),
+    ('max_stack_depth', '2048', 'kB', 'integer', 'superuser'),
+    ('constraint_exclusion', '', None, 'enum', 'user'),
+    ('force_parallel_mode', '1', None, 'enum', 'user'),
+    ('zero_damaged_pages', 'off', None, 'bool', 'superuser'),
+    ('stats_temp_directory', '/tmp', None, 'string', 'sighup'),
+    ('track_commit_timestamp', 'off', None, 'bool', 'postmaster'),
+    ('wal_log_hints', 'on', None, 'bool', 'superuser'),
+    ('hot_standby', 'on', None, 'bool', 'superuser'),
+    ('max_replication_slots', '5', None, 'integer', 'superuser'),
+    ('wal_level', 'logical', None, 'enum', 'superuser'),
+]
+
 
 class MockResponse(object):
 
@@ -133,22 +166,9 @@ class MockCursor(object):
                             ('archive_command', 'my archive command'),
                             ('cluster_name', 'my_cluster')]
         elif sql.startswith('SELECT name, setting'):
-            self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
-                            ('wal_block_size', '8192', None, 'integer', 'internal'),
-                            ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
-                            ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
-                            ('max_connections', '100', None, 'integer', 'postmaster'),
-                            ('max_prepared_transactions', '0', None, 'integer', 'postmaster'),
-                            ('max_worker_processes', '8', None, 'integer', 'postmaster'),
-                            ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'),
-                            ('max_wal_senders', '5', None, 'integer', 'postmaster'),
-                            ('search_path', 'public', None, 'string', 'user'),
-                            ('port', '5433', None, 'integer', 'postmaster'),
-                            ('listen_addresses', '*', None, 'string', 'postmaster'),
-                            ('autovacuum', 'on', None, 'bool', 'sighup'),
-                            ('unix_socket_directories', '/tmp', None, 'string', 'postmaster')]
+            self.results = GET_PG_SETTINGS_RESULT
         elif sql.startswith('SELECT COUNT(*) FROM pg_catalog.pg_settings'):
-            self.results = [(1,)]
+            self.results = [(0,)]
         elif sql.startswith('IDENTIFY_SYSTEM'):
             self.results = [('1', 3, '0/402EEC0', '')]
         elif sql.startswith('TIMELINE_HISTORY '):
@@ -218,11 +238,11 @@ class PostgresInit(unittest.TestCase):
     _PARAMETERS = {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'f.oo': 'bar',
                    'search_path': 'public', 'hot_standby': 'on', 'max_wal_senders': 5,
                    'wal_keep_segments': 8, 'wal_log_hints': 'on', 'max_locks_per_transaction': 64,
-                   'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 0,
+                   'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 200,
                    'track_commit_timestamp': 'off', 'unix_socket_directories': '/tmp',
-                   'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': '',
+                   'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': 'off',
                    'force_parallel_mode': '1', 'constraint_exclusion': '',
-                   'max_stack_depth': 'Z', 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200}
+                   'max_stack_depth': 2048, 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200}
 
     @patch('patroni.psycopg._connect', psycopg_connect)
     @patch('patroni.postgresql.CallbackExecutor', Mock())
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 31454479..8a5c491b 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -5,6 +5,7 @@ import re
 import subprocess
 import time
 
+from copy import deepcopy
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 
 import patroni.psycopg as psycopg
@@ -25,7 +26,8 @@ from patroni.postgresql.validator import (ValidatorFactoryNoType, ValidatorFacto
 from patroni.utils import RetryFailedError
 from threading import Thread, current_thread
 
-from . import BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs
+from . import (BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs,
+               GET_PG_SETTINGS_RESULT)
 
 
 mtime_ret = {}
@@ -559,31 +561,103 @@ class TestPostgresql(BaseTestPostgresql):
 
     @patch('time.sleep', Mock())
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
-    def test_reload_config(self):
-        parameters = self._PARAMETERS.copy()
-        parameters.pop('f.oo')
-        parameters['wal_buffers'] = '512'
-        config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'use_unix_socket_repl': True,
-                  'authentication': {},
-                  'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters}
+    @patch('patroni.postgresql.config.logger.info')
+    @patch('patroni.postgresql.config.logger.warning')
+    def test_reload_config(self, mock_warning, mock_info):
+        config = deepcopy(self.p.config._config)
+
+        # Nothing changed
+        self.p.reload_config(config)
+        mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.')
+        mock_warning.assert_not_called()
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Handle wal_buffers
+        self.p.config._config['parameters']['wal_buffers'] = '512'
         self.p.reload_config(config)
-        parameters['b.ar'] = 'bar'
-        with patch.object(MockCursor, 'fetchall',
-                          Mock(side_effect=[[('wal_block_size', '8191', None, 'integer', 'internal'),
-                                             ('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
-                                             ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
-                                             ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
-                                             ('port', '5433', None, 'integer', 'postmaster')], Exception])):
+        mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.')
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+        config = deepcopy(self.p.config._config)
+
+        # hba/ident_changed
+        config['pg_hba'] = ['']
+        config['pg_ident'] = ['']
+        self.p.reload_config(config)
+        mock_info.assert_called_once_with('Reloading PostgreSQL configuration.')
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Postmaster parameter change (pending_restart)
+        init_max_worker_processes = config['parameters']['max_worker_processes']
+        config['parameters']['max_worker_processes'] *= 2
+        with patch('patroni.postgresql.Postgresql._query', Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)]])):
             self.p.reload_config(config)
-        parameters['autovacuum'] = 'on'
+            self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s (restart might be required)',
+                                                              'max_worker_processes', str(init_max_worker_processes),
+                                                              config['parameters']['max_worker_processes']))
+            self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+            self.assertEqual(self.p.pending_restart, True)
+
+        mock_info.reset_mock()
+
+        # Reset to the initial value without restart
+        config['parameters']['max_worker_processes'] = init_max_worker_processes
+        self.p.reload_config(config)
+        self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'max_worker_processes',
+                                                          init_max_worker_processes * 2,
+                                                          str(config['parameters']['max_worker_processes'])))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # User-defined parameter changed (removed)
+        config['parameters'].pop('f.oo')
         self.p.reload_config(config)
-        parameters['autovacuum'] = 'off'
-        parameters.pop('search_path')
-        config['listen'] = '*:5433'
+        self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'f.oo', 'bar', None))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Non-postmaster parameter change
+        config['parameters']['autovacuum'] = 'off'
         self.p.reload_config(config)
-        parameters['unix_socket_directories'] = '.'
+        self.assertEqual(mock_info.call_args_list[0][0], ("Changed %s from %s to %s", 'autovacuum', 'on', 'off'))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        config['parameters']['autovacuum'] = 'on'
+        mock_info.reset_mock()
+
+        # Remove invalid parameter
+        config['parameters']['invalid'] = 'value'
         self.p.reload_config(config)
-        self.p.config.resolve_connection_addresses()
+        self.assertEqual(mock_warning.call_args_list[0][0],
+                         ('Removing invalid parameter `%s` from postgresql.parameters', 'invalid'))
+        config['parameters'].pop('invalid')
+
+        mock_warning.reset_mock()
+        mock_info.reset_mock()
+
+        # Non-empty result (outside changes) and exception while querying pending_restart parameters
+        with patch('patroni.postgresql.Postgresql._query',
+                   Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)], GET_PG_SETTINGS_RESULT, Exception])):
+            self.p.reload_config(config, True)
+            self.assertEqual(mock_info.call_args_list[0][0], ('Reloading PostgreSQL configuration.',))
+            self.assertEqual(self.p.pending_restart, True)
+
+            # Invalid values, just to increase silly coverage in postgresql.validator.
+            # One day we will have proper tests there.
+            config['parameters']['autovacuum'] = 'of'  # Bool.transform()
+            config['parameters']['vacuum_cost_limit'] = 'smth'  # Number.transform()
+            self.p.reload_config(config, True)
+            self.assertEqual(mock_warning.call_args_list[-1][0][0], 'Exception %r when running query')
 
     def test_resolve_connection_addresses(self):
         self.p.config._config['use_unix_socket'] = self.p.config._config['use_unix_socket_repl'] = True