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/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/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))