Get rid of kubernetes python module (#1586)

The official python kubernetes client contains a lot of auto-generated code and therefore very heavy, but we need only a little fraction of it.
The naive implementation, that covers all API methods we use, takes about 250 LoC, and about half of it is responsible for the handling of configuration files.

Disadvantage: If somebody was using the `patronictl` outside of the pod (on his machine), it might not work anymore (depending on the environment).

features/environment.py

@@ -422,7 +422,7 @@ class KubernetesController(AbstractDcsController):
         os.environ['PATRONI_KUBERNETES_LABELS'] = json.dumps(self._labels)
         os.environ['PATRONI_KUBERNETES_USE_ENDPOINTS'] = 'true'
 
-        from kubernetes import client as k8s_client, config as k8s_config
+        from patroni.dcs.kubernetes import k8s_client, k8s_config
         k8s_config.load_kube_config(context='local')
         self._client = k8s_client
         self._api = self._client.CoreV1Api()

patroni/dcs/kubernetes.py

@@ -1,13 +1,15 @@
-from __future__ import absolute_import
 import datetime
 import functools
 import json
 import logging
+import os
 import socket
+import six
 import sys
 import time
+import urllib3
+import yaml
 
-from kubernetes import client as k8s_client, config as k8s_config, watch as k8s_watch
 from urllib3 import Timeout
 from urllib3.exceptions import HTTPError
 from six.moves.http_client import HTTPException
@@ -15,15 +17,245 @@ from threading import Condition, Lock, Thread
 
 from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, TimelineHistory
 from ..exceptions import DCSError
-from ..utils import deep_compare, keepalive_socket_options, Retry, RetryFailedError, tzutc, USER_AGENT
+from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\
+        Retry, RetryFailedError, tzutc, uri, USER_AGENT
 
 logger = logging.getLogger(__name__)
 
+KUBE_CONFIG_DEFAULT_LOCATION = os.environ.get('KUBECONFIG', '~/.kube/config')
+SERVICE_HOST_ENV_NAME = 'KUBERNETES_SERVICE_HOST'
+SERVICE_PORT_ENV_NAME = 'KUBERNETES_SERVICE_PORT'
+SERVICE_TOKEN_FILENAME = '/var/run/secrets/kubernetes.io/serviceaccount/token'
+SERVICE_CERT_FILENAME = '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt'
+
 
 class KubernetesError(DCSError):
     pass
 
 
+# this function does the same mapping of snake_case => camelCase for > 97% of cases as autogenerated swagger code
+def to_camel_case(value):
+    reserved = {'api', 'apiv3', 'cidr', 'cpu', 'csi', 'id', 'io', 'ip', 'ipc', 'pid', 'tls', 'uri', 'url', 'uuid'}
+    words = value.split('_')
+    return words[0] + ''.join(w.upper() if w in reserved else w.title() for w in words[1:])
+
+
+class K8sConfig(object):
+
+    class ConfigException(Exception):
+        pass
+
+    def __init__(self):
+        self.pool_config = {'maxsize': 10, 'num_pools': 10}  # configuration for urllib3.PoolManager
+        self._make_headers()
+
+    def _make_headers(self, token=None, **kwargs):
+        self._headers = urllib3.make_headers(user_agent=USER_AGENT, **kwargs)
+        if token:
+            self._headers['authorization'] = 'Bearer ' + token
+
+    def load_incluster_config(self):
+        if SERVICE_HOST_ENV_NAME not in os.environ or SERVICE_PORT_ENV_NAME not in os.environ:
+            raise self.ConfigException('Service host/port is not set.')
+        if not os.environ[SERVICE_HOST_ENV_NAME] or not os.environ[SERVICE_PORT_ENV_NAME]:
+            raise self.ConfigException('Service host/port is set but empty.')
+        if not os.path.isfile(SERVICE_CERT_FILENAME):
+            raise self.ConfigException('Service certificate file does not exists.')
+        with open(SERVICE_CERT_FILENAME) as f:
+            if not f.read():
+                raise self.ConfigException('Cert file exists but empty.')
+        if not os.path.isfile(SERVICE_TOKEN_FILENAME):
+            raise self.ConfigException('Service token file does not exists.')
+        with open(SERVICE_TOKEN_FILENAME) as f:
+            token = f.read()
+            if not token:
+                raise self.ConfigException('Token file exists but empty.')
+            self._make_headers(token=token)
+        self.pool_config['ca_certs'] = SERVICE_CERT_FILENAME
+        self._server = uri('https', (os.environ[SERVICE_HOST_ENV_NAME], os.environ[SERVICE_PORT_ENV_NAME]))
+
+    @staticmethod
+    def _get_by_name(config, section, name):
+        for c in config[section + 's']:
+            if c['name'] == name:
+                return c[section]
+
+    def load_kube_config(self, context=None):
+        with open(os.path.expanduser(KUBE_CONFIG_DEFAULT_LOCATION)) as f:
+            config = yaml.safe_load(f)
+
+        context = self._get_by_name(config, 'context', context or config['current-context'])
+        cluster = self._get_by_name(config, 'cluster', context['cluster'])
+        user = self._get_by_name(config, 'user', context['user'])
+
+        self._server = cluster['server'].rstrip('/')
+        if self._server.startswith('https'):
+            self.pool_config.update({v: user[k] for k, v in {'client-certificate': 'cert_file',
+                                                             'client-key': 'key_file'}.items() if k in user})
+            if 'certificate-authority' in cluster:
+                self.pool_config['ca_certs'] = cluster['certificate-authority']
+            self.pool_config['cert_reqs'] = 'CERT_NONE' if cluster.get('insecure-skip-tls-verify') else 'CERT_REQUIRED'
+        if user.get('token'):
+            self._make_headers(token=user['token'])
+        elif 'username' in user and 'password' in user:
+            self._headers = self._make_headers(basic_auth=':'.join((user['username'],  user['password'])))
+
+    @property
+    def server(self):
+        return self._server
+
+    @property
+    def headers(self):
+        return self._headers.copy()
+
+
+class K8sObject(object):
+
+    def __init__(self, kwargs):
+        self._dict = {k: self._wrap(k, v) for k, v in kwargs.items()}
+
+    def get(self, name, default=None):
+        return self._dict.get(name, default)
+
+    def __getattr__(self, name):
+        return self.get(to_camel_case(name))
+
+    @classmethod
+    def _wrap(cls, parent, value):
+        if isinstance(value, dict):
+            # we know that `annotations` and `labels` are dicts and therefore don't want to convert them into K8sObject
+            return value if parent in {'annotations', 'labels'} and \
+                    all(isinstance(v, six.string_types) for v in value.values()) else cls(value)
+        elif isinstance(value, list):
+            return [cls._wrap(None, v) for v in value]
+        else:
+            return value
+
+    def to_dict(self):
+        return self._dict
+
+    def __repr__(self):
+        return json.dumps(self, indent=4, default=lambda o: o.to_dict())
+
+
+class K8sClient(object):
+
+    class rest(object):
+
+        class ApiException(Exception):
+            def __init__(self, status=None, reason=None, http_resp=None):
+                self.status = http_resp.status if http_resp else status
+                self.reason = http_resp.reason if http_resp else reason
+                self.body = http_resp.data if http_resp else None
+                self.headers = http_resp.getheaders() if http_resp else None
+
+            def __str__(self):
+                error_message = "({0})\nReason: {1}\n".format(self.status, self.reason)
+                if self.headers:
+                    error_message += "HTTP response headers: {0}\n".format(self.headers)
+                if self.body:
+                    error_message += "HTTP response body: {0}\n".format(self.body)
+                return error_message
+
+    class ApiClient(object):
+
+        _API_URL_PREFIX = '/api/v1/namespaces/'
+
+        def __init__(self):
+            self.pool_manager = urllib3.PoolManager(**k8s_config.pool_config)
+            self.set_read_timeout(10)
+
+        def set_read_timeout(self, timeout):
+            self._read_timeout = timeout
+
+        @staticmethod
+        def _handle_server_response(response, _preload_content):
+            if response.status not in range(200, 206):
+                raise k8s_client.rest.ApiException(http_resp=response)
+            return K8sObject(json.loads(response.data.decode('utf-8'))) if _preload_content else response
+
+        @staticmethod
+        def _make_headers(headers):
+            ret = k8s_config.headers
+            ret.update(headers or {})
+            return ret
+
+        def request(self, method, path, timeout=None, **kwargs):
+            retries = 0 if timeout else 1
+            if timeout:
+                if isinstance(timeout, six.integer_types + (float,)):
+                    timeout = urllib3.Timeout(total=timeout)
+                elif isinstance(timeout, tuple) and len(timeout) == 2:
+                    timeout = urllib3.Timeout(connect=timeout[0], read=timeout[1])
+            else:
+                timeout = self._read_timeout / 2.0
+                timeout = urllib3.Timeout(connect=max(1, timeout/2.0), total=timeout)
+            kwargs.update(retries=retries, timeout=timeout)
+            return self.pool_manager.request(method, k8s_config.server + path, **kwargs)
+
+        def call_api(self, method, path, headers=None, body=None,
+                     _preload_content=True, _request_timeout=None, **kwargs):
+            headers = self._make_headers(headers)
+            fields = {to_camel_case(k): v for k, v in kwargs.items()}  # resource_version => resourceVersion
+            body = json.dumps(body, default=lambda o: o.to_dict()) if body is not None else None
+
+            response = self.request(method, self._API_URL_PREFIX + path, headers=headers, fields=fields,
+                                    body=body, preload_content=_preload_content, timeout=_request_timeout)
+
+            return self._handle_server_response(response, _preload_content)
+
+    class CoreV1Api(object):
+
+        def __init__(self, api_client=None):
+            self._api_client = api_client or k8s_client.ApiClient()
+
+        def __getattr__(self, func):  # `func` name pattern: (action)_namespaced_(kind)
+            action, kind = func.split('_namespaced_')  # (read|list|create|patch|replace|delete|delete_collection)
+            kind = kind.replace('_', '') + ('s' * int(kind[-1] != 's'))  # plural, single word
+
+            def wrapper(*args, **kwargs):
+                method = {'read': 'GET', 'list': 'GET', 'create': 'POST',
+                          'replace': 'PUT'}.get(action, action.split('_')[0]).upper()
+
+                if action == 'create' or len(args) == 1:  # namespace is a first argument and name in not in arguments
+                    path = '/'.join([args[0], kind])
+                else:  # name, namespace followed by optional body
+                    path = '/'.join([args[1], kind, args[0]])
+
+                headers = {'Content-Type': 'application/strategic-merge-patch+json'} if action == 'patch' else {}
+
+                if len(args) == 3:  # name, namespace, body
+                    body = args[2]
+                elif action == 'create':  # namespace, body
+                    body = args[1]
+                elif action == 'delete':  # name, namespace
+                    body = kwargs.pop('body', None)
+                else:
+                    body = None
+
+                return self._api_client.call_api(method, path, headers, body, **kwargs)
+            return wrapper
+
+    class _K8sObjectTemplate(K8sObject):
+        """The template for objects which we create locally, e.g. k8s_client.V1ObjectMeta & co"""
+        def __init__(self, **kwargs):
+            self._dict = {to_camel_case(k): v for k, v in kwargs.items()}
+
+    def __init__(self):
+        self.__cls_cache = {}
+        self.__cls_lock = Lock()
+
+    def __getattr__(self, name):
+        with self.__cls_lock:
+            if name not in self.__cls_cache:
+                self.__cls_cache[name] = type(name, (self._K8sObjectTemplate,), {})
+        return self.__cls_cache[name]
+
+
+k8s_client = K8sClient()
+k8s_config = K8sConfig()
+
+
 class KubernetesRetriableException(k8s_client.rest.ApiException):
 
     def __init__(self, orig):
@@ -42,10 +274,8 @@ class KubernetesRetriableException(k8s_client.rest.ApiException):
 class CoreV1ApiProxy(object):
 
     def __init__(self, use_endpoints=False):
-        self._api = k8s_client.CoreV1Api()
-        self._api.api_client.user_agent = USER_AGENT
-        self._api.api_client.rest_client.pool_manager.connection_pool_kw['maxsize'] = 10
-        self._request_timeout = None
+        self._api_client = k8s_client.ApiClient()
+        self._core_v1_api = k8s_client.CoreV1Api(self._api_client)
         self._use_endpoints = bool(use_endpoints)
 
     def configure_timeouts(self, loop_wait, retry_timeout, ttl):
@@ -53,19 +283,17 @@ class CoreV1ApiProxy(object):
         # If we didn't received anything after the loop_wait + retry_timeout it is a time
         # to start worrying (send keepalive messages). Finally, the connection should be
         # considered as dead if we received nothing from the socket after the ttl seconds.
-        self._api.api_client.rest_client.pool_manager.connection_pool_kw['socket_options'] = \
+        self._api_client.pool_manager.connection_pool_kw['socket_options'] = \
                 list(keepalive_socket_options(ttl, int(loop_wait + retry_timeout)))
-        self._request_timeout = (1, retry_timeout / 3.0)
+        self._api_client.set_read_timeout(retry_timeout)
 
     def __getattr__(self, func):
         if func.endswith('_kind'):
             func = func[:-4] + ('endpoints' if self._use_endpoints else 'config_map')
 
         def wrapper(*args, **kwargs):
-            if '_request_timeout' not in kwargs:
-                kwargs['_request_timeout'] = self._request_timeout
             try:
-                return getattr(self._api, func)(*args, **kwargs)
+                return getattr(self._core_v1_api, func)(*args, **kwargs)
             except k8s_client.rest.ApiException as e:
                 if e.status in (500, 503, 504) or e.headers and 'retry-after' in e.headers:  # XXX
                     raise KubernetesRetriableException(e)
@@ -97,7 +325,6 @@ class ObjectCache(Thread):
     def __init__(self, dcs, func, retry, condition, name=None):
         Thread.__init__(self)
         self.daemon = True
-        self._api_client = k8s_client.ApiClient()
         self._dcs = dcs
         self._func = func
         self._retry = retry
@@ -155,8 +382,7 @@ class ObjectCache(Thread):
 
         response = self._watch(objects.metadata.resource_version)
         try:
-            for line in k8s_watch.watch.iter_resp_lines(response):
-                event = json.loads(line)
+            for event in iter_response_objects(response):
                 obj = event['object']
                 if obj.get('code') == 410:
                     break
@@ -165,8 +391,7 @@ class ObjectCache(Thread):
                 name = obj['metadata']['name']
 
                 if ev_type in ('ADDED', 'MODIFIED'):
-                    obj = k8s_watch.watch.SimpleNamespace(data=json.dumps(obj))
-                    obj = self._api_client.deserialize(obj, return_type)
+                    obj = K8sObject(obj)
                     success, old_value = self.set(name, obj)
                     if success:
                         new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name))

patroni/utils.py

@@ -1,3 +1,4 @@
+import json.decoder as json_decoder
 import logging
 import os
 import platform
@@ -375,6 +376,27 @@ def uri(proto, netloc, path='', user=None):
     return '{0}://{1}{2}{3}{4}'.format(proto, user, host, port, path)
 
 
+def iter_response_objects(response):
+    prev = ''
+    decoder = json_decoder.JSONDecoder()
+    for chunk in response.read_chunked(decode_content=False):
+        if isinstance(chunk, bytes):
+            chunk = chunk.decode('utf-8')
+        chunk = prev + chunk
+
+        length = len(chunk)
+        idx = json_decoder.WHITESPACE.match(chunk, 0).end()
+        while idx < length:
+            try:
+                message, idx = decoder.raw_decode(chunk, idx)
+            except ValueError:  # malformed or incomplete JSON, unlikely to happen
+                break
+            else:
+                yield message
+                idx = json_decoder.WHITESPACE.match(chunk, idx).end()
+        prev = chunk[idx:]
+
+
 def is_standby_cluster(config):
     # Check whether or not provided configuration describes a standby cluster
     return isinstance(config, dict) and (config.get('host') or config.get('port') or config.get('restore_command'))

requirements.txt

@@ -10,4 +10,3 @@ prettytable>=0.7
 python-dateutil
 psutil>=2.0.0
 cdiff
-kubernetes>=2.0.0,<=10.0.1,!=4.0.*,!=5.0.*

setup.py

@@ -23,7 +23,7 @@ KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
     ' zookeeper exhibitor consul streaming replication kubernetes k8s'
 
 EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'],
-                  'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['kubernetes']}
+                  'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': []}
 COVERAGE_XML = True
 COVERAGE_HTML = False
 
@@ -165,7 +165,7 @@ def setup_package(version):
             continue
         extra = False
         for e, v in EXTRAS_REQUIRE.items():
-            if r.startswith(v[0]):
+            if v and r.startswith(v[0]):
                 EXTRAS_REQUIRE[e] = [r]
                 extra = True
         if not extra:

tests/__init__.py

@@ -22,6 +22,7 @@ class MockResponse(object):
     def __init__(self, status_code=200):
         self.status_code = status_code
         self.content = '{}'
+        self.reason = 'Not Found'
 
     @property
     def data(self):
@@ -35,6 +36,10 @@ class MockResponse(object):
     def getheader(*args):
         return ''
 
+    @staticmethod
+    def getheaders():
+        return {'content-type': 'json'}
+
 
 def requests_get(url, **kwargs):
     members = '[{"id":14855829450254237642,"peerURLs":["http://localhost:2380","http://localhost:7001"],' +\

tests/test_kubernetes.py

@@ -2,10 +2,12 @@ import json
 import time
 import unittest
 
-from mock import Mock, patch
-from patroni.dcs.kubernetes import Kubernetes, KubernetesError, k8s_client, RetryFailedError
+from mock import Mock, mock_open, patch
+from patroni.dcs.kubernetes import Kubernetes, KubernetesError, K8sConfig, K8sObject, RetryFailedError,\
+        k8s_client, k8s_config, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
+from six.moves import builtins
 from threading import Thread
-from . import SleepException
+from . import MockResponse, SleepException
 
 
 def mock_list_namespaced_config_map(*args, **kwargs):
@@ -51,28 +53,99 @@ def mock_namespaced_kind(*args, **kwargs):
     return mock
 
 
+def mock_load_k8s_config(self, *args, **kwargs):
+    self._server = ''
+
+
+class TestK8sConfig(unittest.TestCase):
+
+    def test_load_incluster_config(self):
+        for env in ({}, {SERVICE_HOST_ENV_NAME: '', SERVICE_PORT_ENV_NAME: ''}):
+            with patch('os.environ', env):
+                self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config)
+
+        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
+                patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])),\
+                patch.object(builtins, 'open', Mock(side_effect=[
+                    mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')(),
+                    mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')()])):
+            for _ in range(0, 4):
+                self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config)
+            k8s_config.load_incluster_config()
+            self.assertEqual(k8s_config.server, 'https://a:1')
+
+    def test_load_kube_config(self):
+        config = {
+            "current-context": "local",
+            "contexts": [{"name": "local", "context": {"user": "local", "cluster": "local"}}],
+            "clusters": [{"name": "local", "cluster": {"server": "https://a:1/", "certificate-authority": "a"}}],
+            "users": [{"name": "local", "user": {"username": "a", "password": "b", "client-certificate": "c"}}]
+        }
+        with patch.object(builtins, 'open', mock_open(read_data=json.dumps(config))):
+            k8s_config.load_kube_config()
+            self.assertEqual(k8s_config.server, 'https://a:1')
+            self.assertEqual(k8s_config.pool_config, {'ca_certs': 'a', 'cert_file': 'c', 'cert_reqs': 'CERT_REQUIRED',
+                                                      'maxsize': 10, 'num_pools': 10})
+
+        config["users"][0]["user"]["token"] = "token"
+        with patch.object(builtins, 'open', mock_open(read_data=json.dumps(config))):
+            k8s_config.load_kube_config()
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer token')
+
+
+class TestCoreV1Api(unittest.TestCase):
+
+    @patch.object(K8sConfig, '_server', '', create=True)
+    def setUp(self):
+        self.a = k8s_client.CoreV1Api()
+        self.a._api_client.set_read_timeout(10)
+        self.a._api_client.pool_manager.request = Mock(return_value=MockResponse())
+
+    def test_create_namespaced_service(self):
+        self.assertEqual(str(self.a.create_namespaced_service('default', {}, _request_timeout=2)), '{}')
+
+    def test_list_namespaced_endpoints(self):
+        self.a._api_client.pool_manager.request.return_value.content = '{"items": [1,2,3]}'
+        self.assertIsInstance(self.a.list_namespaced_endpoints('default'), K8sObject)
+
+    def test_patch_namespaced_config_map(self):
+        self.assertEqual(str(self.a.patch_namespaced_config_map('foo', 'default', {}, _request_timeout=(1, 2))), '{}')
+
+    def test_list_namespaced_pod(self):
+        self.a._api_client.pool_manager.request.return_value.status_code = 409
+        self.a._api_client.pool_manager.request.return_value.content = 'foo'
+        try:
+            self.a.list_namespaced_pod('default', label_selector='foo=bar')
+            self.assertFail()
+        except k8s_client.rest.ApiException as e:
+            self.assertTrue('Reason: ' in str(e))
+
+    def test_delete_namespaced_pod(self):
+        self.assertEqual(str(self.a.delete_namespaced_pod('foo', 'default', _request_timeout=(1, 2), body={})), '{}')
+
+
 class BaseTestKubernetes(unittest.TestCase):
 
     @patch('socket.TCP_KEEPIDLE', 4, create=True)
     @patch('socket.TCP_KEEPINTVL', 5, create=True)
     @patch('socket.TCP_KEEPCNT', 6, create=True)
-    @patch('kubernetes.config.load_kube_config', Mock())
-    @patch('kubernetes.client.api_client.ThreadPool', Mock(), create=True)
     @patch.object(Thread, 'start', Mock())
-    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod)
-    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map)
+    @patch.object(K8sConfig, 'load_kube_config', mock_load_k8s_config)
+    @patch.object(K8sConfig, 'load_incluster_config', Mock(side_effect=k8s_config.ConfigException))
+    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod, create=True)
+    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)
     def setUp(self, config=None):
         config = config or {}
         config.update(ttl=30, scope='test', name='p-0', loop_wait=10, retry_timeout=10, labels={'f': 'b'})
         self.k = Kubernetes(config)
         self.assertRaises(AttributeError, self.k._pods._build_cache)
         self.k._pods._is_ready = True
-        self.assertRaises(AttributeError, self.k._kinds._build_cache)
+        self.assertRaises(TypeError, self.k._kinds._build_cache)
         self.k._kinds._is_ready = True
         self.k.get_cluster()
 
 
-@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_namespaced_kind)
+@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_namespaced_kind, create=True)
 class TestKubernetesConfigMaps(BaseTestKubernetes):
 
     @patch('time.time', Mock(side_effect=[1, 10.9, 100]))
@@ -95,13 +168,14 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         self.k.take_leader()
 
     def test_manual_failover(self):
-        with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', Mock(side_effect=RetryFailedError(''))):
+        with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map',
+                          Mock(side_effect=RetryFailedError('')), create=True):
             self.k.manual_failover('foo', 'bar')
 
     def test_set_config_value(self):
         self.k.set_config_value('{}')
 
-    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod')
+    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', create=True)
     def test_touch_member(self, mock_patch_namespaced_pod):
         mock_patch_namespaced_pod.return_value.metadata.resource_version = '10'
         self.k.touch_member({'role': 'replica'})
@@ -119,7 +193,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         self.k.cancel_initialization()
 
     @patch.object(k8s_client.CoreV1Api, 'delete_collection_namespaced_config_map',
-                  Mock(side_effect=k8s_client.rest.ApiException(403, '')))
+                  Mock(side_effect=k8s_client.rest.ApiException(403, '')), create=True)
     def test_delete_cluster(self):
         self.k.delete_cluster()
 
@@ -134,11 +208,11 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
 
 class TestKubernetesEndpoints(BaseTestKubernetes):
 
-    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints)
+    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True)
     def setUp(self, config=None):
         super(TestKubernetesEndpoints, self).setUp({'use_endpoints': True, 'pod_ip': '10.0.0.0'})
 
-    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints')
+    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
     def test_update_leader(self, mock_patch_namespaced_endpoints):
         self.assertIsNotNone(self.k.update_leader('123'))
         args = mock_patch_namespaced_endpoints.call_args[0]
@@ -148,11 +222,11 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
         self.k._kinds._object_cache['test'].metadata.annotations['leader'] = 'p-1'
         self.assertFalse(self.k.update_leader('123'))
 
-    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind)
+    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True)
     def test_update_leader_with_restricted_access(self):
         self.assertIsNotNone(self.k.update_leader('123', True))
 
-    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints')
+    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
     def test__update_leader_with_retry(self, mock_patch):
         mock_patch.side_effect = k8s_client.rest.ApiException(502, '')
         self.assertFalse(self.k.update_leader('123'))
@@ -168,14 +242,15 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
             self.assertIsNotNone(self.k._update_leader_with_retry({}, '1', []))
 
     @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints',
-                  Mock(side_effect=[k8s_client.rest.ApiException(500, ''), k8s_client.rest.ApiException(502, '')]))
+                  Mock(side_effect=[k8s_client.rest.ApiException(500, ''),
+                                    k8s_client.rest.ApiException(502, '')]), create=True)
     def test_delete_sync_state(self):
         self.assertFalse(self.k.delete_sync_state())
 
-    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', mock_namespaced_kind)
-    @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', mock_namespaced_kind)
+    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', mock_namespaced_kind, create=True)
+    @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', mock_namespaced_kind, create=True)
     @patch.object(k8s_client.CoreV1Api, 'create_namespaced_service',
-                  Mock(side_effect=[True, False, k8s_client.rest.ApiException(500, '')]))
+                  Mock(side_effect=[True, False, k8s_client.rest.ApiException(500, '')]), create=True)
     def test__create_config_service(self):
         self.assertIsNotNone(self.k.patch_or_create_config({'foo': 'bar'}))
         self.assertIsNotNone(self.k.patch_or_create_config({'foo': 'bar'}))
@@ -184,7 +259,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
 
 class TestCacheBuilder(BaseTestKubernetes):
 
-    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map)
+    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)
     @patch('patroni.dcs.kubernetes.ObjectCache._watch')
     def test__build_cache(self, mock_response):
         mock_response.return_value.read_chunked.return_value = [json.dumps(
@@ -195,7 +270,7 @@ class TestCacheBuilder(BaseTestKubernetes):
                 'name': self.k.config_path, 'resourceVersion': '3'}}}
         ) + '\n' + json.dumps(
             {'type': 'MDIFIED', 'object': {'metadata': {'name': self.k.config_path}}}
-        ) + '\n' + json.dumps({'object': {'code': 410}}) + '\n').encode('utf-8')]
+        ) + '\n').encode('utf-8'), b'{"object":{', b'"code":410}}\n']
         self.k._kinds._build_cache()
 
     @patch('patroni.dcs.kubernetes.logger.error', Mock(side_effect=SleepException))