Repository: zalando/patroni
Pull request #2893 (2023-10-02 01:48:37)

Add warning of removing user creation
add a warning of removing user creation.


Patch:
diff --git a/features/environment.py b/features/environment.py
index 1f36eb0c..3e0ad3b1 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -692,7 +692,7 @@ class ZooKeeperController(AbstractExternalDcsController):
         self._client = kazoo.client.KazooClient()
 
     def process_name(self):
-        return "zookeeper"
+        return "java .*zookeeper"
 
     def query(self, key, scope='batman', group=None):
         import kazoo.exceptions
diff --git a/features/recovery.feature b/features/recovery.feature
index 809f7fb9..7839e26f 100644
--- a/features/recovery.feature
+++ b/features/recovery.feature
@@ -14,6 +14,8 @@ Feature: recovery
     Then I receive a response code 200
     And I receive a response role master
     And I receive a response timeline 1
+    And "members/postgres0" key in DCS has state=running after 12 seconds
+    And replication works from postgres0 to postgres1 after 15 seconds
 
   Scenario: check immediate failover when master_start_timeout=0
     Given I issue a PATCH request to http://127.0.0.1:8008/config with {"master_start_timeout": 0}
diff --git a/patroni/api.py b/patroni/api.py
index 911ea574..e1b71337 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -451,7 +451,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
         Write an HTTP response with JSON content based on the output of :func:`~patroni.utils.cluster_as_json`, with
         HTTP status ``200`` and the JSON representation of the cluster topology.
         """
-        cluster = self.server.patroni.dcs.get_cluster(True)
+        cluster = self.server.patroni.dcs.get_cluster()
         global_config = self.server.patroni.config.get_global_config(cluster)
 
         response = cluster_as_json(cluster, global_config)
@@ -690,7 +690,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
         """
         request = self._read_json_content()
         if request:
-            cluster = self.server.patroni.dcs.get_cluster(True)
+            cluster = self.server.patroni.dcs.get_cluster()
             if not (cluster.config and cluster.config.modify_version):
                 return self.send_error(503)
             data = cluster.config.data.copy()
@@ -1166,7 +1166,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         patroni = self.server.patroni
         if patroni.postgresql.citus_handler.is_coordinator() and patroni.ha.is_leader():
-            cluster = patroni.dcs.get_cluster(True)
+            cluster = patroni.dcs.get_cluster()
             patroni.postgresql.citus_handler.handle_event(cluster, request)
         self.write_response(200, 'OK')
 
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index d71d41ad..1a516cca 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -1548,9 +1548,6 @@ class AbstractDCS(abc.ABC):
                 primary and exception raised, instance would be demoted.
         """
 
-    def _bypass_caches(self) -> None:
-        """Used only in Zookeeper."""
-
     def __get_patroni_cluster(self, path: Optional[str] = None) -> Cluster:
         """Low level method to load a :class:`Cluster` object from DCS.
 
@@ -1593,14 +1590,13 @@ class AbstractDCS(abc.ABC):
                   dict.
         """
         groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader)
-        if isinstance(groups, Cluster):  # Zookeeper could return a cached version
-            cluster = groups
-        else:
-            cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
-            cluster.workers.update(groups)
+        if TYPE_CHECKING:  # pragma: no cover
+            assert isinstance(groups, dict)
+        cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
+        cluster.workers.update(groups)
         return cluster
 
-    def get_cluster(self, force: bool = False) -> Cluster:
+    def get_cluster(self) -> Cluster:
         """Retrieve an appropriate cached or fresh view of DCS.
 
         .. note::
@@ -1609,12 +1605,8 @@ class AbstractDCS(abc.ABC):
 
             Returns either a Citus or Patroni implementation of :class:`Cluster` depending on availability.
 
-        :param force: a value of ``True`` will override Zookeeper caching features.
-
         :returns:
         """
-        if force:
-            self._bypass_caches()
         try:
             cluster = self._get_citus_cluster() if self.is_citus_coordinator() else self.__get_patroni_cluster()
         except Exception:
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index 649c1ac5..3093ba06 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -116,10 +116,7 @@ class ZooKeeper(AbstractDCS):
                                           timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
                                           sleep_func=time.sleep), command_retry=KazooRetry(max_delay=1, max_tries=-1,
                                           deadline=config['retry_timeout'], sleep_func=time.sleep), **kwargs)
-        self._client.add_listener(self.session_listener)
 
-        self._fetch_cluster: bool = True
-        self._fetch_status: bool = True
         self.__last_member_data: Optional[Dict[str, Any]] = None
 
         self._orig_kazoo_connect = self._client._connection._connect
@@ -142,18 +139,9 @@ class ZooKeeper(AbstractDCS):
         ret = self._orig_kazoo_connect(*args)
         return max(self.loop_wait - 2, 2) * 1000, ret[1]
 
-    def session_listener(self, state: str) -> None:
-        if state in [KazooState.SUSPENDED, KazooState.LOST]:
-            self.cluster_watcher(None)
-
-    def status_watcher(self, event: Optional[WatchedEvent]) -> None:
-        self._fetch_status = True
-        self.event.set()
-
-    def cluster_watcher(self, event: Optional[WatchedEvent]) -> None:
-        self._fetch_cluster = True
-        if not event or event.state != KazooState.CONNECTED or event.path.startswith(self.client_path('')):
-            self.status_watcher(event)
+    def _watcher(self, event: WatchedEvent) -> None:
+        if event.state != KazooState.CONNECTED or event.path.startswith(self.client_path('')):
+            self.event.set()
 
     def reload_config(self, config: Union['Config', Dict[str, Any]]) -> None:
         self.set_retry_timeout(config['retry_timeout'])
@@ -202,75 +190,66 @@ class ZooKeeper(AbstractDCS):
             return None
 
     def get_status(self, path: str, leader: Optional[Leader]) -> Status:
-        watch = self.status_watcher if not leader or leader.name != self._name else None
-
-        status = self.get_node(path + self._STATUS, watch)
+        status = self.get_node(path + self._STATUS)
         if not status:
-            status = self.get_node(path + self._LEADER_OPTIME, watch)
-        if status:
-            self._fetch_status = False
+            status = self.get_node(path + self._LEADER_OPTIME)
         return Status.from_node(status and status[0])
 
     @staticmethod
     def member(name: str, value: str, znode: ZnodeStat) -> Member:
         return Member.from_node(znode.version, name, znode.ephemeralOwner, value)
 
-    def get_children(self, key: str, watch: Optional[Callable[[WatchedEvent], None]] = None) -> List[str]:
+    def get_children(self, key: str) -> List[str]:
         try:
-            return self._client.get_children(key, watch)
+            return self._client.get_children(key)
         except NoNodeError:
             return []
 
     def load_members(self, path: str) -> List[Member]:
         members: List[Member] = []
-        for member in self.get_children(path + self._MEMBERS, self.cluster_watcher):
+        for member in self.get_children(path + self._MEMBERS):
             data = self.get_node(path + self._MEMBERS + member)
             if data is not None:
                 members.append(self.member(member, *data))
         return members
 
     def _cluster_loader(self, path: str) -> Cluster:
-        self._fetch_cluster = False
-        self.event.clear()
-        nodes = set(self.get_children(path, self.cluster_watcher))
-        if not nodes:
-            self._fetch_cluster = True
+        nodes = set(self.get_children(path))
 
         # get initialize flag
         initialize = (self.get_node(path + self._INITIALIZE) or [None])[0] if self._INITIALIZE in nodes else None
 
         # get global dynamic configuration
-        config = self.get_node(path + self._CONFIG, watch=self.cluster_watcher) if self._CONFIG in nodes else None
+        config = self.get_node(path + self._CONFIG, watch=self._watcher) if self._CONFIG in nodes else None
         config = config and ClusterConfig.from_node(config[1].version, config[0], config[1].mzxid)
 
         # get timeline history
-        history = self.get_node(path + self._HISTORY, watch=self.cluster_watcher) if self._HISTORY in nodes else None
+        history = self.get_node(path + self._HISTORY) if self._HISTORY in nodes else None
         history = history and TimelineHistory.from_node(history[1].mzxid, history[0])
 
         # get synchronization state
-        sync = self.get_node(path + self._SYNC, watch=self.cluster_watcher) if self._SYNC in nodes else None
+        sync = self.get_node(path + self._SYNC) if self._SYNC in nodes else None
         sync = SyncState.from_node(sync and sync[1].version, sync and sync[0])
 
         # get list of members
         members = self.load_members(path) if self._MEMBERS[:-1] in nodes else []
 
         # get leader
-        leader = self.get_node(path + self._LEADER) if self._LEADER in nodes else None
+        leader = self.get_node(path + self._LEADER, watch=self._watcher) if self._LEADER in nodes else None
         if leader:
             member = Member(-1, leader[0], None, {})
             member = ([m for m in members if m.name == leader[0]] or [member])[0]
             leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
-            self._fetch_cluster = member.version == -1
 
         # get last known leader lsn and slots
         status = self.get_status(path, leader)
 
         # failover key
-        failover = self.get_node(path + self._FAILOVER, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
+        failover = self.get_node(path + self._FAILOVER) if self._FAILOVER in nodes else None
         failover = failover and Failover.from_node(failover[1].version, failover[0])
 
         # get failsafe topology
-        failsafe = self.get_node(path + self._FAILSAFE, watch=self.cluster_watcher) if self._FAILSAFE in nodes else None
+        failsafe = self.get_node(path + self._FAILSAFE) if self._FAILSAFE in nodes else None
         try:
             failsafe = json.loads(failsafe[0]) if failsafe else None
         except Exception:
@@ -279,45 +258,20 @@ class ZooKeeper(AbstractDCS):
         return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
-        fetch_cluster = False
         ret: Dict[int, Cluster] = {}
-        for node in self.get_children(path, self.cluster_watcher):
+        for node in self.get_children(path):
             if citus_group_re.match(node):
                 ret[int(node)] = self._cluster_loader(path + node + '/')
-                fetch_cluster = fetch_cluster or self._fetch_cluster
-        self._fetch_cluster = fetch_cluster
         return ret
 
     def _load_cluster(
             self, path: str, loader: Callable[[str], Union[Cluster, Dict[int, Cluster]]]
     ) -> Union[Cluster, Dict[int, Cluster]]:
-        cluster = self.cluster if path == self._base_path + '/' else None
-        if self._fetch_cluster or cluster is None:
-            try:
-                cluster = self._client.retry(loader, path)
-            except Exception:
-                logger.exception('get_cluster')
-                self.cluster_watcher(None)
-                raise ZooKeeperError('ZooKeeper in not responding properly')
-        # The /status ZNode was updated or doesn't exist
-        elif self._fetch_status and not self._fetch_cluster or not cluster.last_lsn \
-                or cluster.has_permanent_slots(self._name) and not cluster.slots:
-            # If current node is the leader just clear the event without fetching anything (we are updating the /status)
-            if cluster.leader and cluster.leader.name == self._name:
-                self.event.clear()
-            else:
-                try:
-                    status = self.get_status(self.client_path(''), cluster.leader)
-                    self.event.clear()
-                    new_cluster: List[Any] = list(cluster)
-                    new_cluster[3] = status
-                    cluster = Cluster(*new_cluster)
-                except Exception:
-                    pass
-        return cluster
-
-    def _bypass_caches(self) -> None:
-        self._fetch_cluster = True
+        try:
+            return self._client.retry(loader, path)
+        except Exception:
+            logger.exception('get_cluster')
+            raise ZooKeeperError('ZooKeeper in not responding properly')
 
     def _create(self, path: str, value: bytes, retry: bool = False, ephemeral: bool = False) -> bool:
         try:
@@ -334,7 +288,6 @@ class ZooKeeper(AbstractDCS):
         try:
             self._client.retry(self._client.create, self.leader_path, self._name.encode('utf-8'),
                                makepath=True, ephemeral=True)
-            self.cluster_watcher(None)  # the next _load_cluster() call must read from ZooKeeper.
             return True
         except (ConnectionClosedError, RetryFailedError) as e:
             raise ZooKeeperError(e)
@@ -381,20 +334,9 @@ class ZooKeeper(AbstractDCS):
         member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
         member_data = self.__last_member_data or member and member.data
         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):
+            # We want delete the member ZNode if our session doesn't match with session id on our member key
+            if self._client.client_id is not None and member.session != self._client.client_id[0]:
+                logger.warning('Recreating the member ZNode due to ownership mismatch')
                 try:
                     self._client.delete_async(self.member_path).get(timeout=1)
                 except NoNodeError:
@@ -493,7 +435,10 @@ class ZooKeeper(AbstractDCS):
         return self.set_sync_state_value("{}", version) is not False
 
     def watch(self, leader_version: Optional[int], timeout: float) -> bool:
-        ret = super(ZooKeeper, self).watch(leader_version, timeout + 0.5)
-        if ret and not self._fetch_status:
-            self._fetch_cluster = True
-        return ret or self._fetch_cluster
+        if leader_version:
+            timeout += 0.5
+
+        try:
+            return super(ZooKeeper, self).watch(leader_version, timeout)
+        finally:
+            self.event.clear()
diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py
index 26025e43..8a9b0fba 100644
--- a/patroni/postgresql/bootstrap.py
+++ b/patroni/postgresql/bootstrap.py
@@ -400,6 +400,9 @@ BEGIN
 END;$$""".format(f, quote_ident(rewind['username'], postgresql.connection()))
                         postgresql.query(sql)
 
+                if config.get('users'):
+                    logger.warning('User creation via "bootstrap.users" will be removed in v4.0.0')
+
                 for name, value in (config.get('users') or {}).items():
                     if all(name != a.get('username') for a in (superuser, replication, rewind)):
                         self.create_or_update_role(name, value.get('password'), value.get('options', []))


Test patch:
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index 184fda05..3ce3ea75 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -1,13 +1,13 @@
 import select
 import unittest
 
-from kazoo.client import KazooClient, KazooState
+from kazoo.client import KazooClient
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.handlers.threading import SequentialThreadingHandler
-from kazoo.protocol.states import KeeperState, ZnodeStat
+from kazoo.protocol.states import KeeperState, WatchedEvent, ZnodeStat
 from kazoo.retry import RetryFailedError
 from mock import Mock, PropertyMock, patch
-from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient, \
+from patroni.dcs.zookeeper import Cluster, PatroniKazooClient, \
     PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
 
 
@@ -152,9 +152,6 @@ class TestZooKeeper(unittest.TestCase):
                              'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10,
                              'set_acls': {'CN=principal2': ['ALL']}})
 
-    def test_session_listener(self):
-        self.zk.session_listener(KazooState.SUSPENDED)
-
     def test_reload_config(self):
         self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10})
         self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 5})
@@ -176,15 +173,6 @@ class TestZooKeeper(unittest.TestCase):
         self.zk._cluster_loader(self.zk.client_path(''))
 
     def test_get_cluster(self):
-        cluster = self.zk.get_cluster(True)
-        self.assertIsInstance(cluster.leader, Leader)
-        self.zk.status_watcher(None)
-        self.zk.get_cluster()
-        self.zk.touch_member({'foo': 'foo'})
-        self.zk._name = 'bar'
-        self.zk.status_watcher(None)
-        with patch.object(ZooKeeper, 'get_node', Mock(side_effect=Exception)):
-            self.zk.get_cluster()
         cluster = self.zk.get_cluster()
         self.assertEqual(cluster.last_lsn, 500)
 
@@ -295,3 +283,7 @@ class TestZooKeeper(unittest.TestCase):
 
     def test_set_history_value(self):
         self.zk.set_history_value('{}')
+
+    def test_watcher(self):
+        self.zk._watcher(WatchedEvent('', '', ''))
+        self.assertTrue(self.zk.watch(1, 1))