Repository: zalando/patroni
Pull request #2876 (2023-09-21 11:47:39)

Read GUC's values when joining running Postgres
If restarted in pause Patroni was discarding `synchronous_standby_names` from `postgresql.conf` because in the internal cache this values was set to `None`. As a result synchronous replication transitioned to a broken state, with no synchronous replicas according to the `synchronous_standby_names` and Patroni not selecting/setting the new synchronous replicas (another bug).

To solve the problem of broken initial state and to avoid similar issues with other GUC's we will read GUC's value if Patroni is joining running Postgres.

coveralls: ## Pull Request Test Coverage Report for [Build 6261224662](https://coveralls.io/builds/62808734)

* **20** of **20**   **(100.0%)**  changed or added relevant lines in **3** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.863%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/62808734/badge)](https://coveralls.io/builds/62808734) |
| :-- | --: |
| Change from base [Build 6250333541](https://coveralls.io/builds/62784539): |  0.0% |
| Covered Lines: | 13109 |
| Relevant Lines: | 13127 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/features/patroni_api.feature b/features/patroni_api.feature
index 624d3271..ff06f3c9 100644
--- a/features/patroni_api.feature
+++ b/features/patroni_api.feature
@@ -68,6 +68,7 @@ Scenario: check API requests for the primary-replica pair in the pause mode
 	When I kill postmaster on postgres1
 	And I issue a GET request to http://127.0.0.1:8009/replica
 	Then I receive a response code 503
+	And "members/postgres1" key in DCS has state=stopped after 10 seconds
 	When I run patronictl.py restart batman postgres1 --force
 	Then I receive a response returncode 0
 	Then replication works from postgres0 to postgres1 after 20 seconds
@@ -76,7 +77,7 @@ Scenario: check API requests for the primary-replica pair in the pause mode
 	Then I receive a response code 200
 	And I receive a response state running
 	And I receive a response role replica
-	When I run patronictl.py reinit batman postgres1 --force
+	When I run patronictl.py reinit batman postgres1 --force --wait
 	Then I receive a response returncode 0
 	And I receive a response output "Success: reinitialize for member postgres1"
 	And postgres1 role is the secondary after 30 seconds
diff --git a/patroni/__main__.py b/patroni/__main__.py
index 2b318a67..7d56172b 100644
--- a/patroni/__main__.py
+++ b/patroni/__main__.py
@@ -106,6 +106,8 @@ class Patroni(AbstractPatroniDaemon, Tags):
 
     def ensure_unique_name(self) -> None:
         """A helper method to prevent splitbrain from operator naming error."""
+        from urllib.parse import urlparse
+        from urllib3.connection import HTTPConnection
         from patroni.dcs import Member
 
         cluster = self.dcs.get_cluster()
@@ -115,9 +117,12 @@ class Patroni(AbstractPatroniDaemon, Tags):
         if not isinstance(member, Member):
             return
         try:
-            _ = self.request(member, endpoint="/liveness", timeout=3)
-            logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
-            sys.exit(1)
+            parts = urlparse(member.api_url)
+            if isinstance(parts.hostname, str):
+                connection = HTTPConnection(parts.hostname, port=parts.port or 80, timeout=3)
+                connection.connect()
+                logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
+                sys.exit(1)
         except Exception:
             return
 
diff --git a/patroni/config.py b/patroni/config.py
index 1650934d..fee2147a 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -490,7 +490,8 @@ class Config(object):
             elif not is_local:
                 validator = ConfigHandler.CMDLINE_OPTIONS[name][1]
                 if validator(value):
-                    pg_params[name] = int(value) if isinstance(validator, IntValidator) else value
+                    int_val = parse_int(value) if isinstance(validator, IntValidator) else None
+                    pg_params[name] = int_val if isinstance(int_val, int) else value
                 else:
                     logger.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
                                    name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])
diff --git a/patroni/config_generator.py b/patroni/config_generator.py
index 956e18d2..0269c49a 100644
--- a/patroni/config_generator.py
+++ b/patroni/config_generator.py
@@ -287,7 +287,7 @@ class RunningClusterConfigGenerator(AbstractConfigGenerator):
 
         :param cur: connection cursor to use.
         """
-        cur.execute("SELECT name, current_setting(name) FROM pg_settings "
+        cur.execute("SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings "
                     "WHERE context <> 'internal' "
                     "AND source IN ('configuration file', 'command line', 'environment variable') "
                     "AND category <> 'Write-Ahead Log / Recovery Target' "
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index 6390aa2e..863d7b40 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -379,21 +379,28 @@ class ZooKeeper(AbstractDCS):
         cluster = self.cluster
         member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
         member_data = self.__last_member_data or member and member.data
-        #  We want to notify leader if some important fields in the member key changed by removing ZNode
-        if member and (self._client.client_id is not None and member.session != self._client.client_id[0]
-                       or not (member_data and deep_compare(member_data.get('tags', {}), data.get('tags', {}))
-                               and (member_data.get('state') == data.get('state')
-                                    or 'running' not in (member_data.get('state'), data.get('state')))
-                               and member_data.get('version') == data.get('version')
-                               and member_data.get('checkpoint_after_promote')
-                               == data.get('checkpoint_after_promote'))):
-            try:
-                self._client.delete_async(self.member_path).get(timeout=1)
-            except NoNodeError:
-                pass
-            except Exception:
-                return False
-            member = None
+        if member and member_data:
+            is_leader = data.get('role') in ('master', 'primary', 'standby_leader')
+            checkpoint_after_promote_changed = member_data.get('checkpoint_after_promote') \
+                != data.get('checkpoint_after_promote')
+            state_running_changed = member_data.get('state') != data.get('state') \
+                and 'running' in (member_data.get('state'), data.get('state'))
+            tags_changed = not deep_compare(member_data.get('tags', {}), data.get('tags', {}))
+
+            # We want delete the member ZNode if:
+            # - our session doesn't match with session id on our member key; or
+            # - we want to notify leader if some important fields in the member key changed; or
+            # - if we are the leader and want to notify replicas about checkpoint_after_promote;
+            if self._client.client_id is not None and member.session != self._client.client_id[0] \
+                    or is_leader and checkpoint_after_promote_changed \
+                    or not is_leader and (state_running_changed or tags_changed):
+                try:
+                    self._client.delete_async(self.member_path).get(timeout=1)
+                except NoNodeError:
+                    pass
+                except Exception:
+                    return False
+                member = None
 
         encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8')
         if member and member_data:
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index af48886f..cab33e8f 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -119,17 +119,28 @@ class Postgresql(object):
         # Last known running process
         self._postmaster_proc = None
 
-        if self.is_running():  # we are "joining" already running postgres
-            self.set_state('running')
+        if self.is_running():
+            # If we found postmaster process we need to figure out whether postgres is accepting connections
+            self.set_state('starting')
+            self.check_startup_state_changed()
+
+        if self.state == 'running':  # we are "joining" already running postgres
+            # we know that PostgreSQL is accepting connections and can read some GUC's from pg_settings
+            self.config.load_current_server_parameters()
+
             self.set_role('master' if self.is_primary() else 'replica')
-            # postpone writing postgresql.conf for 12+ because recovery parameters are not yet known
-            if self.major_version < 120000 or self.is_primary():
-                self.config.write_postgresql_conf()
+
             hba_saved = self.config.replace_pg_hba()
             ident_saved = self.config.replace_pg_ident()
-            if hba_saved or ident_saved:
+
+            if self.major_version < 120000 or self.role in ('master', 'primary'):
+                # If PostgreSQL is running as a primary or we run PostgreSQL that is older than 12 we can
+                # call reload_config() once again (the first call happened in the ConfigHandler constructor),
+                # so that it can figure out if config files should be updated and pg_ctl reload executed.
+                self.config.reload_config(config, sighup=bool(hba_saved or ident_saved))
+            elif hba_saved or ident_saved:
                 self.reload()
-        elif self.role in ('master', 'primary'):
+        elif not self.is_running() and self.role in ('master', 'primary'):
             self.set_role('demoted')
 
     @property
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index a918a2cf..40d87f35 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -326,14 +326,22 @@ class ConfigHandler(object):
                                         .format(self._pgpass))
         self._passfile = None
         self._passfile_mtime = None
-        self._synchronous_standby_names = None
         self._postmaster_ctime = None
         self._current_recovery_params: Optional[CaseInsensitiveDict] = None
         self._config = {}
         self._recovery_params = CaseInsensitiveDict()
-        self._server_parameters: CaseInsensitiveDict
+        self._server_parameters: CaseInsensitiveDict = CaseInsensitiveDict()
         self.reload_config(config)
 
+    def load_current_server_parameters(self) -> None:
+        """Read GUC's values from ``pg_settings`` when Patroni is joining the the postgres that is already running."""
+        exclude = [name.lower() for name, value in self.CMDLINE_OPTIONS.items() if value[1] == _false_validator] \
+            + [name.lower() for name in self._RECOVERY_PARAMETERS]
+        self._server_parameters = CaseInsensitiveDict({r[0]: r[1] for r in self._postgresql.query(
+            "SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings"
+            " WHERE (source IN ('command line', 'environment variable') OR sourcefile = %s)"
+            " AND pg_catalog.lower(name) != ALL(%s)", self._postgresql_conf, exclude)})
+
     def setup_server_parameters(self) -> None:
         self._server_parameters = self.get_server_parameters(self._config)
         self._adjust_recovery_parameters()
@@ -922,14 +930,15 @@ class ConfigHandler(object):
         listen_addresses, port = split_host_port(config['listen'], 5432)
         parameters.update(cluster_name=self._postgresql.scope, listen_addresses=listen_addresses, port=str(port))
         if not self._postgresql.global_config or self._postgresql.global_config.is_synchronous_mode:
-            if self._synchronous_standby_names is None:
+            synchronous_standby_names = self._server_parameters.get('synchronous_standby_names')
+            if synchronous_standby_names is None:
                 if self._postgresql.global_config and self._postgresql.global_config.is_synchronous_mode_strict\
                         and self._postgresql.role in ('master', 'primary', 'promoted'):
                     parameters['synchronous_standby_names'] = '*'
                 else:
                     parameters.pop('synchronous_standby_names', None)
             else:
-                parameters['synchronous_standby_names'] = self._synchronous_standby_names
+                parameters['synchronous_standby_names'] = synchronous_standby_names
 
         # Handle hot_standby <-> replica rename
         if parameters.get('wal_level') == ('hot_standby' if self._postgresql.major_version >= 90600 else 'replica'):
@@ -1150,12 +1159,11 @@ class ConfigHandler(object):
     def set_synchronous_standby_names(self, value: Optional[str]) -> Optional[bool]:
         """Updates synchronous_standby_names and reloads if necessary.
         :returns: True if value was updated."""
-        if value != self._synchronous_standby_names:
+        if value != self._server_parameters.get('synchronous_standby_names'):
             if value is None:
                 self._server_parameters.pop('synchronous_standby_names', None)
             else:
                 self._server_parameters['synchronous_standby_names'] = value
-            self._synchronous_standby_names = value
             if self._postgresql.state == 'running':
                 self.write_postgresql_conf()
                 self._postgresql.reload()


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index 5240cec6..32172926 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -117,7 +117,7 @@ class MockCursor(object):
             self.results = [(False, 2)]
         elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'):
             self.results = [(datetime.datetime.now(tzutc),)]
-        elif sql.startswith('SELECT name, current_setting(name) FROM pg_settings'):
+        elif sql.startswith('SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings'):
             self.results = [('data_directory', 'data'),
                             ('hba_file', os.path.join('data', 'pg_hba.conf')),
                             ('ident_file', os.path.join('data', 'pg_ident.conf')),
@@ -137,6 +137,11 @@ class MockCursor(object):
                             ('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'),
@@ -248,6 +253,7 @@ class PostgresInit(unittest.TestCase):
 
 class BaseTestPostgresql(PostgresInit):
 
+    @patch('time.sleep', Mock())
     def setUp(self):
         super(BaseTestPostgresql, self).setUp()
 
diff --git a/tests/test_config.py b/tests/test_config.py
index bd8d0a90..f0a780bc 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -155,6 +155,7 @@ class TestConfig(unittest.TestCase):
         expected_params = {
             'f.oo': 'bar',  # not in ConfigHandler.CMDLINE_OPTIONS
             'max_connections': 100,  # IntValidator
+            'wal_keep_size': '128MB',  # IntValidator
             'wal_level': 'hot_standby',  # EnumValidator
         }
         input_params = deepcopy(expected_params)
diff --git a/tests/test_patroni.py b/tests/test_patroni.py
index 0385731c..df59677d 100644
--- a/tests/test_patroni.py
+++ b/tests/test_patroni.py
@@ -40,7 +40,7 @@ class MockFrozenImporter(object):
 @patch('time.sleep', Mock())
 @patch('subprocess.call', Mock(return_value=0))
 @patch('patroni.psycopg.connect', psycopg_connect)
-@patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
+@patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
 @patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -64,7 +64,7 @@ class TestPatroni(unittest.TestCase):
             self.assertRaises(SystemExit, _main)
 
     @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()]))
-    @patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
+    @patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
     @patch('sys.frozen', Mock(return_value=True), create=True)
     @patch.object(HTTPServer, '__init__', Mock())
     @patch.object(etcd.Client, 'read', etcd_read)
@@ -108,6 +108,7 @@ class TestPatroni(unittest.TestCase):
     @patch('os.getpid')
     @patch('multiprocessing.Process')
     @patch('patroni.__main__.patroni_main', Mock())
+    @patch('sys.argv', ['patroni.py', 'postgres0.yml'])
     def test_patroni_main(self, mock_process, mock_getpid):
         mock_getpid.return_value = 2
         _main()
@@ -233,8 +234,8 @@ class TestPatroni(unittest.TestCase):
         )
         with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)):
             # If the api of the running node cannot be reached, this implies unique name
-            with patch.object(self.p, 'request', Mock(side_effect=ConnectionError)):
+            with patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=ConnectionError)):
                 self.assertIsNone(self.p.ensure_unique_name())
             # Only if the api of the running node is reachable do we throw an error
-            with patch.object(self.p, 'request', Mock()):
+            with patch('urllib3.connection.HTTPConnection.connect', Mock()):
                 self.assertRaises(SystemExit, self.p.ensure_unique_name)
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index e2fad6ba..6ae69e85 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -721,6 +721,7 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertEqual(self.p.get_primary_timeline(), 1)
 
     @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='replica'))
+    @patch.object(Postgresql, 'is_running', Mock(return_value=False))
     @patch.object(Bootstrap, 'running_custom_bootstrap', PropertyMock(return_value=True))
     @patch.object(Postgresql, 'controldata', Mock(return_value={'max_connections setting': '200',
                                                                 'max_worker_processes setting': '20',
@@ -964,6 +965,7 @@ class TestPostgresql2(BaseTestPostgresql):
     @patch('patroni.postgresql.CallbackExecutor', Mock())
     @patch.object(Postgresql, 'get_major_version', Mock(return_value=140000))
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def setUp(self):
         super(TestPostgresql2, self).setUp()
 
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index 45aeac10..184fda05 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -276,6 +276,7 @@ class TestZooKeeper(unittest.TestCase):
         self.assertTrue(self.zk.delete_cluster())
 
     def test_watch(self):
+        self.zk.event.wait = Mock()
         self.zk.watch(None, 0)
         self.zk.event.is_set = Mock(return_value=True)
         self.zk._fetch_status = False