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

@@ -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,23 +560,9 @@ class ObjectCache(Thread):
         with self._object_cache_lock:
             return self._object_cache.get(name)
 
-    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
-
+    def _process_event(self, event):
         ev_type = event['type']
+        obj = event['object']
         name = obj['metadata']['name']
 
         if ev_type in ('ADDED', 'MODIFIED'):
@@ -586,10 +574,9 @@ class ObjectCache(Thread):
             success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
             new_value = None
         else:
-                    logger.warning('Unexpected event type: %s', ev_type)
-                    continue
+            return logger.warning('Unexpected event type: %s', ev_type)
 
-                if success and return_type != 'V1Pod':
+        if success and obj.get('kind') != 'Pod':
             if old_value:
                 old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
 
@@ -602,19 +589,70 @@ class ObjectCache(Thread):
             # 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()
+        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()
+
+        try:
+            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
+        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

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