From 3c1b274ab72f705faca9c2d47f641bc343eb734f Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 10 Jul 2023 09:19:43 +0200 Subject: [PATCH] Use quorum read in patronictl if it is possible (#2730) implementations and terminologis are DCS specific: - Etcd v2 calls is `quorum` read - Etcd v3 calls it `linearizable` (vs `serializable`) - Consul calls it `consistent` Following DCS don't offer this feature: - ZooKeeper calls it linearizable, but reads are sequentially consistent - Raft - no quorum reads are possible ATM - Kubernetes - uses Etcd under the hood, but provides no API to choose read consistency level Close https://github.com/zalando/patroni/issues/1199 --- patroni/dcs/consul.py | 8 ++++++-- patroni/dcs/etcd.py | 4 ++-- patroni/dcs/etcd3.py | 11 ++++++----- patroni/dcs/kubernetes.py | 4 ++-- 4 files changed, 16 insertions(+), 11 deletions(-) diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index a62e7747..73755a41 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -400,8 +400,12 @@ class Consul(AbstractDCS): return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe) + @property + def _consistency(self) -> str: + return 'consistent' if self._ctl else self._client.consistency + def _cluster_loader(self, path: str) -> Cluster: - _, results = self.retry(self._client.kv.get, path, recurse=True) + _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency) if results is None: raise NotFound nodes = {} @@ -412,7 +416,7 @@ class Consul(AbstractDCS): return self._cluster_from_nodes(nodes) def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: - _, results = self.retry(self._client.kv.get, path, recurse=True) + _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency) clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict) for node in results or []: key = node['Key'][len(path):].split('/', 1) diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 5f667771..94e546ac 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -725,13 +725,13 @@ class Etcd(AbstractEtcd): return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe) def _cluster_loader(self, path: str) -> Cluster: - result = self.retry(self._client.read, path, recursive=True) + result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl) nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves} return self._cluster_from_nodes(result.etcd_index, nodes) def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict) - result = self.retry(self._client.read, path, recursive=True) + result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl) for node in result.leaves: key = node.key[len(result.key):].lstrip('/').split('/', 1) if len(key) == 2 and citus_group_re.match(key[0]): diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index efb955e0..32d00359 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -326,14 +326,14 @@ class Etcd3Client(AbstractEtcdClientWithFailover): return retry(e) @_handle_auth_errors - def range(self, key: str, range_end: Union[bytes, str, None] = None, + def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True, retry: Optional[Retry] = None) -> Dict[str, Any]: params = build_range_request(key, range_end) - params['serializable'] = True # For better performance. We can tolerate stale reads. + params['serializable'] = serializable # For better performance. We can tolerate stale reads return self.call_rpc('/kv/range', params, retry) - def prefix(self, key: str, retry: Optional[Retry] = None) -> Dict[str, Any]: - return self.range(key, prefix_range_end(key), retry) + def prefix(self, key: str, serializable: bool = True, retry: Optional[Retry] = None) -> Dict[str, Any]: + return self.range(key, prefix_range_end(key), serializable, retry) @_handle_auth_errors def lease_grant(self, ttl: int, retry: Optional[Retry] = None) -> str: @@ -594,7 +594,8 @@ class PatroniEtcd3Client(Etcd3Client): self._wait_cache(self.read_timeout) ret = self._kv_cache.copy() else: - ret = self._etcd3.retry(self.prefix, path).get('kvs', []) + serializable = not getattr(self._etcd3, '_ctl') # use linearizable for patronictl + ret = self._etcd3.retry(self.prefix, path, serializable).get('kvs', []) for node in ret: node.update({'key': base64_decode(node['key']), 'value': base64_decode(node.get('value', '')), diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 498cfd0f..f770ac64 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -766,7 +766,7 @@ class Kubernetes(AbstractDCS): k8s_config.load_kube_config(context=config.get('context', 'kind-kind')) pod_ip = config.get('pod_ip') - self.__ips: List[str] = [] if config.get('patronictl') or not isinstance(pod_ip, str) else [pod_ip] + self.__ips: List[str] = [] if self._ctl or not isinstance(pod_ip, str) else [pod_ip] self.__ports: List[K8sObject] = [] ports: List[Dict[str, Any]] = config.get('ports', [{}]) for p in ports: @@ -774,7 +774,7 @@ class Kubernetes(AbstractDCS): port.update({n: p[n] for n in ('name', 'protocol') if p.get(n)}) self.__ports.append(k8s_client.V1EndpointPort(**port)) - bypass_api_service = not config.get('patronictl') and config.get('bypass_api_service') + bypass_api_service = not self._ctl and config.get('bypass_api_service') self._api = CoreV1ApiProxy(config.get('use_endpoints'), bypass_api_service) self._should_create_config_service = self._api.use_endpoints self.reload_config(config)