From 881bba9e1cd081f7cd4c340f9a6ad977daa92c58 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 15 May 2020 18:04:59 +0200 Subject: [PATCH] Sync HA loops of all pods in one cluster (#1515) There is no expire mechanism available on K8s, therefore we implement soft leader lock, i.e. every pod is "watching" for changes of the leader object and when there are no changes during the TTL it starts leader race. Before we switched to LIST+WATCH approach in #1189 and #1276, we only watched for the leader object and every time it was updated, the main thread of the HA loop was waking up. As a result, all replica pods were synchronized, and starting the leader race more or less at the same time. The new approach made all pods "unsynchronized" and the biggest downside of it - it takes `ttl + loop_wait` in the worst case to detect the leader failure. This commit makes all pods in one cluster to sync HA loops again based on updates of the leader object. --- patroni/dcs/kubernetes.py | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 15f4463d..3325bed0 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -98,7 +98,7 @@ def catch_kubernetes_errors(func): class ObjectCache(Thread): - def __init__(self, dcs, func, retry, condition): + def __init__(self, dcs, func, retry, condition, name=None): Thread.__init__(self) self.daemon = True self._api_client = k8s_client.ApiClient() @@ -106,6 +106,7 @@ class ObjectCache(Thread): self._func = func self._retry = retry self._condition = condition + self._name = name # name of this pod self._is_ready = False self._object_cache = {} self._object_cache_lock = Lock() @@ -180,9 +181,14 @@ class ObjectCache(Thread): if old_value: old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name)) - if old_value != new_value and \ - (name != self._dcs.config_path or old_value is not None and new_value is not None): + value_changed = old_value != new_value and \ + (name != self._dcs.config_path or old_value is not None and new_value is not None) + + if value_changed: logger.debug('%s changed from %s to %s', name, old_value, new_value) + + # Do not wake up HA loop if we run as leader and received leader object update event + if value_changed or name == self._dcs.leader_path and self._name != new_value: self._dcs.event.set() finally: with self._condition: @@ -251,7 +257,7 @@ class Kubernetes(AbstractDCS): kinds_func = functools.partial(self._api.list_namespaced_kind, self._namespace, label_selector=self._label_selector) - self._kinds = ObjectCache(self, kinds_func, self._retry, self._condition) + self._kinds = ObjectCache(self, kinds_func, self._retry, self._condition, self._name) def retry(self, *args, **kwargs): return self._retry.copy()(*args, **kwargs) @@ -582,6 +588,6 @@ class Kubernetes(AbstractDCS): return True try: - return super(Kubernetes, self).watch(None, timeout) + return super(Kubernetes, self).watch(None, timeout + 0.5) finally: self.event.clear()