From a68692a3e4d3c372a88d6ea1de085052931a7111 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 17 Jul 2020 08:31:58 +0200 Subject: [PATCH] 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 | 2 +- patroni/dcs/kubernetes.py | 259 +++++++++++++++++++++++++++++++++++--- patroni/utils.py | 22 ++++ requirements.txt | 1 - setup.py | 4 +- tests/__init__.py | 5 + tests/test_kubernetes.py | 119 ++++++++++++++---- 7 files changed, 369 insertions(+), 43 deletions(-) diff --git a/features/environment.py b/features/environment.py index 851c41a1..ab65768d 100644 --- a/features/environment.py +++ b/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() diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index a4630578..7a27376a 100644 --- a/patroni/dcs/kubernetes.py +++ b/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)) diff --git a/patroni/utils.py b/patroni/utils.py index 34c70034..c97d63a3 100644 --- a/patroni/utils.py +++ b/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')) diff --git a/requirements.txt b/requirements.txt index b4e07f7e..debdeaa0 100644 --- a/requirements.txt +++ b/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.* diff --git a/setup.py b/setup.py index 45a4b034..f536fa49 100644 --- a/setup.py +++ b/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: diff --git a/tests/__init__.py b/tests/__init__.py index 3cff8183..4958f152 100644 --- a/tests/__init__.py +++ b/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"],' +\ diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 03185348..7e8f0d1b 100644 --- a/tests/test_kubernetes.py +++ b/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))