From ad3d95341085755b45768a2efad60660571e53ac Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 19 May 2022 15:24:20 +0200 Subject: [PATCH] K8s: reset watchers if PATCH fails with 409 (#2283) High CPU load on Etcd nodes and K8s API servers created a very strange situation. A few clusters were running without a leader and the pod which is ahead of others was failing to take a leader lock because updates were failing with HTTP response code `409` (`resource_version` mismatch). Effectively that means that TCP connections to K8s master nodes were alive (in the opposite case tcp keepalives would have resolved it), but no `UPDATE` events were arriving via these connections, resulting in the stale cache of the cluster in memory. The only good way to prevent this situation is to intercept 409 HTTP responses and terminate existing TCP connections used for watches. Now a few words about implementation. Unfortunately, watch threads are waiting in the read() call most of the time and there is no good way to interrupt them. But, the `socket.shutdown()` seems to do this job. We already used this trick in the Etcd3 implementation. This approach will help to mitigate the issue of not having a leader, but at the same time replicas might still end up with the stale cluster state cached and in the worst case will not stream from the leader. Non-streaming replicas are less dangerous and could be covered by monitoring and partially mitigated by correctly configured `archive_command` and `restore_command`. --- patroni/dcs/kubernetes.py | 125 ++++++++++++++++++++++++++------------ tests/test_kubernetes.py | 21 ++++++- 2 files changed, 104 insertions(+), 42 deletions(-) diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 059bd369..3f1c1e51 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -518,6 +518,8 @@ class ObjectCache(Thread): self._condition = condition self._name = name # name of this pod self._is_ready = False + self._response = None # needs to be accessible from the `kill_stream()` method + self._response_lock = Lock() # protect the `self._response` from concurrent access self._object_cache = {} self._object_cache_lock = Lock() self._annotations_map = {self._dcs.leader_path: self._dcs._LEADER, self._dcs.config_path: self._dcs._CONFIG} @@ -558,63 +560,99 @@ class ObjectCache(Thread): with self._object_cache_lock: return self._object_cache.get(name) + def _process_event(self, event): + ev_type = event['type'] + obj = event['object'] + name = obj['metadata']['name'] + + if ev_type in ('ADDED', 'MODIFIED'): + obj = K8sObject(obj) + success, old_value = self.set(name, obj) + if success: + new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name)) + elif ev_type == 'DELETED': + success, old_value = self.delete(name, obj['metadata']['resourceVersion']) + new_value = None + else: + return logger.warning('Unexpected event type: %s', ev_type) + + if success and obj.get('kind') != 'Pod': + if old_value: + old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name)) + + 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() + + @staticmethod + def _finish_response(response): + try: + response.close() + finally: + response.release_conn() + + def _do_watch(self, resource_version): + with self._response_lock: + self._response = None + response = self._watch(resource_version) + with self._response_lock: + if self._response is None: + self._response = response + + if not self._response: + return self._finish_response(response) + + for event in iter_response_objects(response): + if event['object'].get('code') == 410: + break + self._process_event(event) + def _build_cache(self): objects = self._list() - return_type = 'V1' + objects.kind[:-4] with self._object_cache_lock: self._object_cache = {item.metadata.name: item for item in objects.items} with self._condition: self._is_ready = True self._condition.notify() - response = self._watch(objects.metadata.resource_version) try: - for event in iter_response_objects(response): - obj = event['object'] - if obj.get('code') == 410: - break - - ev_type = event['type'] - name = obj['metadata']['name'] - - if ev_type in ('ADDED', 'MODIFIED'): - obj = K8sObject(obj) - success, old_value = self.set(name, obj) - if success: - new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name)) - elif ev_type == 'DELETED': - success, old_value = self.delete(name, obj['metadata']['resourceVersion']) - new_value = None - else: - logger.warning('Unexpected event type: %s', ev_type) - continue - - if success and return_type != 'V1Pod': - if old_value: - old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name)) - - 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() + self._do_watch(objects.metadata.resource_version) finally: with self._condition: self._is_ready = False - response.close() - response.release_conn() + with self._response_lock: + response, self._response = self._response, None + if response: + self._finish_response(response) + + def kill_stream(self): + sock = None + with self._response_lock: + if self._response: + try: + sock = self._response.connection.sock + except Exception: + sock = None + else: + self._response = False + if sock: + try: + sock.shutdown(socket.SHUT_RDWR) + sock.close() + except Exception as e: + logger.debug('Error on socket.shutdown: %r', e) def run(self): while True: try: self._build_cache() except Exception as e: - with self._condition: - self._is_ready = False logger.error('ObjectCache.run %r', e) def is_ready(self): @@ -889,7 +927,14 @@ class Kubernetes(AbstractDCS): def patch_or_create(self, name, annotations, resource_version=None, patch=False, retry=True, ips=None): if retry is True: retry = self.retry - return self._patch_or_create(name, annotations, resource_version, patch, retry, ips) + try: + return self._patch_or_create(name, annotations, resource_version, patch, retry, ips) + except k8s_client.rest.ApiException as e: + if e.status == 409 and resource_version: # Conflict in resource_version + # Terminate watchers, it could be a sign that K8s API is in a failed state + self._kinds.kill_stream() + self._pods.kill_stream() + raise e def patch_or_create_config(self, annotations, resource_version=None, patch=False, retry=True): # SCOPE-config endpoint requires corresponding service otherwise it might be "cleaned" by k8s master diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 982edb5d..fbdd9b67 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -4,7 +4,7 @@ import socket import time import unittest -from mock import Mock, mock_open, patch +from mock import Mock, PropertyMock, mock_open, patch from patroni.dcs.kubernetes import k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\ K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\ Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME @@ -235,7 +235,9 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k.manual_failover('foo', 'bar') def test_set_config_value(self): - self.k.set_config_value('{}') + with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', + Mock(side_effect=k8s_client.rest.ApiException(409, '')), create=True): + self.k.set_config_value('{}', 1) @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', create=True) def test_touch_member(self, mock_patch_namespaced_pod): @@ -345,3 +347,18 @@ class TestCacheBuilder(BaseTestKubernetes): def test__list(self): self.k._pods._func = Mock(side_effect=Exception) self.assertRaises(Exception, self.k._pods._list) + + @patch('patroni.dcs.kubernetes.ObjectCache._watch', Mock(return_value=None)) + def test__do_watch(self): + self.assertRaises(AttributeError, self.k._kinds._do_watch, '1') + + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True) + @patch('patroni.dcs.kubernetes.ObjectCache._watch') + def test_kill_stream(self, mock_watch): + self.k._kinds.kill_stream() + mock_watch.return_value.read_chunked.return_value = [] + mock_watch.return_value.connection.sock.close.side_effect = Exception + self.k._kinds._do_watch('1') + self.k._kinds.kill_stream() + type(mock_watch.return_value).connection = PropertyMock(side_effect=Exception) + self.k._kinds.kill_stream()