Repository: zalando/patroni
Pull request #2884 (2023-09-25 14:05:26)

Take into account current role when deciding on removal of member ZNode
Patroni doesn't watch on all changes of member keys in order to not create too much load on ZooKeeper, but only subscribes to changes (ZNodes added or deleted) in the `/member` directory. Therefore when some important fields in the value are updated we remove and recreate ZNode in order to notify the leader or other members.

The leader should remove the member key only when the `checkpoint_after_promote` value is changed and replicas when the `state` is changed to/from `running`.

We don't care about the `version` field, because Patroni version can't be changed without restart, what will case ZooKeeper `session_id` to change it anyway.

This fix hopefully will reduce failures of behave tests on GH Actions.

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

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

---



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

---
##### 💛  - [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/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:


Test patch:
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_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