gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Populate references and nodename in subsets addresses (#1591)

Browse Source 
It makes subsets to exactly look like they were populated by the service with label selector and would help with https://github.com/zalando/postgres-operator/issues/340#issuecomment-587001109

Unit-tests are refactored to minimize amount of mocks.
This commit is contained in:
Alexander Kukushkin
2020-06-16 12:56:20 +02:00
committed by GitHub
parent 623b594539
commit ee4bf79c11
2 changed files with 138 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/dcs/kubernetes.py
+70 -46
View File
@@ -46,7 +46,7 @@ class CoreV1ApiProxy(object):
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._use_endpoints = use_endpoints
self._use_endpoints = bool(use_endpoints)
def configure_timeouts(self, loop_wait, retry_timeout, ttl):
# Normally every loop_wait seconds we should have receive something from the socket.
@@ -80,6 +80,10 @@ class CoreV1ApiProxy(object):
raise
return wrapper
@property
def use_endpoints(self):
return self._use_endpoints
def catch_kubernetes_errors(func):
def wrapper(*args, **kwargs):
@@ -229,18 +233,16 @@ class Kubernetes(AbstractDCS):
except k8s_config.ConfigException:
k8s_config.load_kube_config(context=config.get('context', 'local'))
self.__subsets = None
use_endpoints = config.get('use_endpoints') and (config.get('patronictl') or 'pod_ip' in config)
if use_endpoints:
addresses = [k8s_client.V1EndpointAddress(ip='127.0.0.1' if config.get('patronictl') else config['pod_ip'])]
ports = []
self.__my_pod = None
self.__ips = [] if config.get('patronictl') else [config.get('pod_ip')]
self.__ports = []
for p in config.get('ports', [{}]):
port = {'port': int(p.get('port', '5432'))}
port.update({n: p[n] for n in ('name', 'protocol') if p.get(n)})
ports.append(k8s_client.V1EndpointPort(**port))
self.__subsets = [k8s_client.V1EndpointSubset(addresses=addresses, ports=ports)]
self._should_create_config_service = True
self._api = CoreV1ApiProxy(use_endpoints)
self.__ports.append(k8s_client.V1EndpointPort(**port))
self._api = CoreV1ApiProxy(config.get('use_endpoints'))
self._should_create_config_service = self._api.use_endpoints
self.reload_config(config)
self._leader_observed_record = {}
self._leader_observed_time = None
@@ -267,7 +269,7 @@ class Kubernetes(AbstractDCS):
@property
def leader_path(self):
return self._base_path[1:] if self.__subsets else super(Kubernetes, self).leader_path
return self._base_path[1:] if self._api.use_endpoints else super(Kubernetes, self).leader_path
def set_ttl(self, ttl):
ttl = int(ttl)
@@ -305,7 +307,9 @@ class Kubernetes(AbstractDCS):
with self._condition:
self._wait_caches()
members = [self.member(pod) for pod in self._pods.copy().values()]
pods = self._pods.copy()
self.__my_pod = pods.get(self._name)
members = [self.member(pod) for pod in pods.values()]
nodes = self._kinds.copy()
config = nodes.get(self.config_path)
@@ -328,7 +332,8 @@ class Kubernetes(AbstractDCS):
leader = nodes.get(self.leader_path)
metadata = leader and leader.metadata
self._leader_resource_version = metadata.resource_version if metadata else None
self._leader_observed_subsets = leader.subsets if self.__subsets and leader and leader.subsets else []
self._leader_observed_subsets = leader.subsets \
if self._api.use_endpoints and leader and leader.subsets else []
annotations = metadata and metadata.annotations or {}
# get last leader operation
@@ -377,50 +382,70 @@ class Kubernetes(AbstractDCS):
return p1.name == p2.name and p1.port == p2.port and (p1.protocol or 'TCP') == (p2.protocol or 'TCP')
@staticmethod
def subsets_changed(last_observed_subsets, subsets):
def subsets_changed(last_observed_subsets, ip, ports):
"""
>>> Kubernetes.subsets_changed([], [])
False
>>> Kubernetes.subsets_changed([], [k8s_client.V1EndpointSubset()])
>>> Kubernetes.subsets_changed([], None, [])
True
>>> s1 = [k8s_client.V1EndpointSubset(addresses=[k8s_client.V1EndpointAddress(ip='1.2.3.4')])]
>>> s2 = [k8s_client.V1EndpointSubset(addresses=[k8s_client.V1EndpointAddress(ip='1.2.3.5')])]
>>> Kubernetes.subsets_changed(s1, s2)
>>> ip = '1.2.3.4'
>>> a = [k8s_client.V1EndpointAddress(ip=ip)]
>>> s = [k8s_client.V1EndpointSubset(addresses=a)]
>>> Kubernetes.subsets_changed(s, '1.2.3.5', [])
True
>>> a = [k8s_client.V1EndpointAddress(ip='1.2.3.4')]
>>> s1 = [k8s_client.V1EndpointSubset(addresses=a, ports=[k8s_client.V1EndpointPort(protocol='TCP', port=1)])]
>>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[k8s_client.V1EndpointPort(port=5432)])]
>>> Kubernetes.subsets_changed(s1, s2)
>>> s = [k8s_client.V1EndpointSubset(addresses=a, ports=[k8s_client.V1EndpointPort(protocol='TCP', port=1)])]
>>> Kubernetes.subsets_changed(s, '1.2.3.4', [k8s_client.V1EndpointPort(port=5432)])
True
>>> p1 = k8s_client.V1EndpointPort(name='port1', port=1)
>>> p2 = k8s_client.V1EndpointPort(name='port2', port=2)
>>> p3 = k8s_client.V1EndpointPort(name='port3', port=3)
>>> s1 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p1, p2])]
>>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p2, p3])]
>>> Kubernetes.subsets_changed(s1, s2)
>>> s = [k8s_client.V1EndpointSubset(addresses=a, ports=[p1, p2])]
>>> Kubernetes.subsets_changed(s, ip, [p2, p3])
True
>>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p2, p1])]
>>> Kubernetes.subsets_changed(s1, s2)
>>> Kubernetes.subsets_changed(s, ip, [p2, p1])
False
"""
if len(last_observed_subsets) != len(subsets):
if len(last_observed_subsets) != 1:
return True
if subsets == []:
return False
if len(last_observed_subsets[0].addresses or []) != 1 or \
last_observed_subsets[0].addresses[0].ip != subsets[0].addresses[0].ip or \
len(last_observed_subsets[0].ports) != len(subsets[0].ports):
last_observed_subsets[0].addresses[0].ip != ip or \
len(last_observed_subsets[0].ports) != len(ports):
return True
if len(subsets[0].ports) == 1:
return not Kubernetes.compare_ports(last_observed_subsets[0].ports[0], subsets[0].ports[0])
if len(ports) == 1:
return not Kubernetes.compare_ports(last_observed_subsets[0].ports[0], ports[0])
observed_ports = {p.name: p for p in last_observed_subsets[0].ports}
for p in subsets[0].ports:
for p in ports:
if p.name not in observed_ports or not Kubernetes.compare_ports(p, observed_ports.pop(p.name)):
return True
return False
def __target_ref(self, leader_ip, pod):
# we want to re-use existing target_ref if possible
for subset in self._leader_observed_subsets:
for address in subset.addresses or []:
if address.ip == leader_ip and address.target_ref and address.target_ref.name == self._name:
return address.target_ref
return k8s_client.V1ObjectReference(kind='Pod', uid=pod.metadata.uid, namespace=self._namespace,
name=self._name, resource_version=pod.metadata.resource_version)
def _map_subsets(self, endpoints, ips):
if not ips:
# We want to have subsets empty
if self._leader_observed_subsets:
endpoints['subsets'] = []
return
pod = self.__my_pod
leader_ip = ips[0] or pod and pod.status.pod_ip
# don't touch subsets if our (leader) ip is unknown or subsets is valid
if leader_ip and self.subsets_changed(self._leader_observed_subsets, leader_ip, self.__ports):
kwargs = {'hostname': pod.spec.hostname, 'node_name': pod.spec.node_name,
'target_ref': self.__target_ref(leader_ip, pod)} if pod else {}
address = k8s_client.V1EndpointAddress(ip=leader_ip, **kwargs)
endpoints['subsets'] = [k8s_client.V1EndpointSubset(addresses=[address], ports=self.__ports)]
@catch_kubernetes_errors
def patch_or_create(self, name, annotations, resource_version=None, patch=False, retry=True, subsets=None):
def patch_or_create(self, name, annotations, resource_version=None, patch=False, retry=True, ips=None):
metadata = {'namespace': self._namespace, 'name': name, 'labels': self._labels, 'annotations': annotations}
if patch or resource_version:
if resource_version is not None:
@@ -432,10 +457,9 @@ class Kubernetes(AbstractDCS):
metadata['annotations'] = {k: v for k, v in metadata['annotations'].items() if v is not None}
metadata = k8s_client.V1ObjectMeta(**metadata)
if subsets is not None and self.__subsets:
if ips is not None and self._api.use_endpoints:
endpoints = {'metadata': metadata}
if self.subsets_changed(self._leader_observed_subsets, subsets):
endpoints['subsets'] = subsets
self._map_subsets(endpoints, ips)
body = k8s_client.V1Endpoints(**endpoints)
else:
body = k8s_client.V1ConfigMap(metadata=metadata)
@@ -446,7 +470,7 @@ class Kubernetes(AbstractDCS):
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
if self.__subsets and not patch and not resource_version:
if self._api.use_endpoints and not patch and not resource_version:
self._should_create_config_service = True
self._create_config_service()
ret = self.patch_or_create(self.config_path, annotations, resource_version, patch, retry)
@@ -479,9 +503,9 @@ class Kubernetes(AbstractDCS):
if last_operation:
annotations[self._OPTIME] = last_operation
subsets = [] if access_is_restricted else self.__subsets
ips = [] if access_is_restricted else self.__ips
ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, subsets=subsets)
ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, ips=ips)
if ret:
self._leader_resource_version = ret.metadata.resource_version
return ret
@@ -501,8 +525,8 @@ class Kubernetes(AbstractDCS):
else:
annotations['acquireTime'] = self._leader_observed_record.get('acquireTime') or now
annotations['transitions'] = str(transitions)
subsets = [] if self.__subsets else None
ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, subsets=subsets)
ips = [] if self._api.use_endpoints else None
ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, ips=ips)
if ret:
self._leader_resource_version = ret.metadata.resource_version
else:
@@ -543,7 +567,7 @@ class Kubernetes(AbstractDCS):
'annotations': {'status': json.dumps(data, separators=(',', ':'))}}
body = k8s_client.V1Pod(metadata=k8s_client.V1ObjectMeta(**metadata))
ret = self._api.patch_namespaced_pod(self._name, self._namespace, body)
if self.__subsets and self._should_create_config_service:
if self._should_create_config_service:
self._create_config_service()
return ret
tests/test_kubernetes.py
+64 -63
View File
@@ -8,7 +8,7 @@ from threading import Thread
from . import SleepException
def mock_list_namespaced_config_map(self, *args, **kwargs):
def mock_list_namespaced_config_map(*args, **kwargs):
metadata = {'resource_version': '1', 'labels': {'f': 'b'}, 'name': 'test-config',
'annotations': {'initialize': '123', 'config': '{}'}}
items = [k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))]
@@ -22,73 +22,72 @@ def mock_list_namespaced_config_map(self, *args, **kwargs):
return k8s_client.V1ConfigMapList(metadata=metadata, items=items, kind='ConfigMapList')
def mock_list_namespaced_pod(self, *args, **kwargs):
metadata = k8s_client.V1ObjectMeta(resource_version='1', name='p-0', annotations={'status': '{}'})
items = [k8s_client.V1Pod(metadata=metadata)]
def mock_list_namespaced_endpoints(*args, **kwargs):
target_ref = k8s_client.V1ObjectReference(kind='Pod', resource_version='10', name='p-0',
namespace='default', uid='964dfeae-e79b-4476-8a5a-1920b5c2a69d')
address0 = k8s_client.V1EndpointAddress(ip='10.0.0.0', target_ref=target_ref)
address1 = k8s_client.V1EndpointAddress(ip='10.0.0.1')
port = k8s_client.V1EndpointPort(port=5432, name='postgresql', protocol='TCP')
subset = k8s_client.V1EndpointSubset(addresses=[address1, address0], ports=[port])
metadata = k8s_client.V1ObjectMeta(resource_version='1', labels={'f': 'b'}, name='test',
annotations={'optime': '1234', 'leader': 'p-0', 'ttl': '30s'})
endpoint = k8s_client.V1Endpoints(subsets=[subset], metadata=metadata)
metadata = k8s_client.V1ObjectMeta(resource_version='1')
return k8s_client.V1EndpointsList(metadata=metadata, items=[endpoint], kind='V1EndpointsList')
def mock_list_namespaced_pod(*args, **kwargs):
metadata = k8s_client.V1ObjectMeta(resource_version='1', name='p-0', annotations={'status': '{}'},
uid='964dfeae-e79b-4476-8a5a-1920b5c2a69d')
status = k8s_client.V1PodStatus(pod_ip='10.0.0.0')
spec = k8s_client.V1PodSpec(hostname='p-0', node_name='kind-control-plane', containers=[])
items = [k8s_client.V1Pod(metadata=metadata, status=status, spec=spec)]
return k8s_client.V1PodList(items=items, kind='PodList')
def mock_config_map(*args, **kwargs):
def mock_namespaced_kind(*args, **kwargs):
mock = Mock()
mock.metadata.resource_version = '2'
return mock
@patch('socket.TCP_KEEPIDLE', 4, create=True)
@patch('socket.TCP_KEEPINTVL', 5, create=True)
@patch('socket.TCP_KEEPCNT', 6, create=True)
@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_config_map)
@patch.object(k8s_client.CoreV1Api, 'create_namespaced_config_map', mock_config_map)
@patch('kubernetes.client.api_client.ThreadPool', Mock(), create=True)
@patch.object(Thread, 'start', Mock())
class TestKubernetes(unittest.TestCase):
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.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map)
@patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod)
@patch('kubernetes.client.api_client.ThreadPool', Mock(), create=True)
@patch.object(Thread, 'start', Mock())
def setUp(self):
self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0',
'loop_wait': 10, 'retry_timeout': 10, 'labels': {'f': 'b'}})
@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)
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.k._kinds._is_ready = True
self.k.get_cluster()
@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_namespaced_kind)
class TestKubernetesConfigMaps(BaseTestKubernetes):
@patch('time.time', Mock(side_effect=[1, 10.9, 100]))
def test__wait_caches(self):
self.k._pods._is_ready = False
with self.k._condition:
self.assertRaises(RetryFailedError, self.k._wait_caches)
@patch('time.time', Mock(return_value=time.time() + 100))
def test_get_cluster(self):
with patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map), \
patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod), \
patch('time.time', Mock(return_value=time.time() + 31)):
self.k.get_cluster()
with patch.object(Kubernetes, '_wait_caches', Mock(side_effect=Exception)):
self.assertRaises(KubernetesError, self.k.get_cluster)
@patch('kubernetes.config.load_kube_config', Mock())
@patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', Mock())
def test_update_leader(self):
k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
self.assertIsNotNone(k.update_leader('123'))
@patch('kubernetes.config.load_kube_config', Mock())
@patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', Mock())
def test_update_leader_with_restricted_access(self):
k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
self.assertIsNotNone(k.update_leader('123', True))
def test_take_leader(self):
self.k.take_leader()
self.k._leader_observed_record['leader'] = 'test'
@@ -123,14 +122,6 @@ class TestKubernetes(unittest.TestCase):
def test_delete_cluster(self):
self.k.delete_cluster()
@patch('kubernetes.config.load_kube_config', Mock())
@patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints',
Mock(side_effect=[k8s_client.rest.ApiException(502, ''), k8s_client.rest.ApiException(500, '')]))
def test_delete_sync_state(self):
k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
self.assertFalse(k.delete_sync_state())
def test_watch(self):
self.k.set_ttl(10)
self.k.watch(None, 0)
@@ -139,31 +130,41 @@ class TestKubernetes(unittest.TestCase):
def test_set_history_value(self):
self.k.set_history_value('{}')
@patch('kubernetes.config.load_kube_config', Mock())
@patch('patroni.dcs.kubernetes.ObjectCache', Mock())
@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', Mock(return_value=True))
@patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', Mock())
class TestKubernetesEndpoints(BaseTestKubernetes):
@patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints)
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')
def test_update_leader(self, mock_patch_namespaced_endpoints):
self.assertIsNotNone(self.k.update_leader('123'))
args = mock_patch_namespaced_endpoints.call_args[0]
self.assertEqual(args[2].subsets[0].addresses[0].target_ref.resource_version, '10')
self.k._leader_observed_subsets = []
self.assertIsNotNone(self.k.update_leader('123'))
@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind)
def test_update_leader_with_restricted_access(self):
self.assertIsNotNone(self.k.update_leader('123', True))
@patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints',
Mock(side_effect=[k8s_client.rest.ApiException(502, ''), k8s_client.rest.ApiException(500, '')]))
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, 'create_namespaced_service',
Mock(side_effect=[True, False, k8s_client.rest.ApiException(500, '')]))
def test__create_config_service(self):
k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
self.assertIsNotNone(k.patch_or_create_config({'foo': 'bar'}))
self.assertIsNotNone(k.patch_or_create_config({'foo': 'bar'}))
k.touch_member({'state': 'running', 'role': 'replica'})
self.assertIsNotNone(self.k.patch_or_create_config({'foo': 'bar'}))
self.assertIsNotNone(self.k.patch_or_create_config({'foo': 'bar'}))
self.k.touch_member({'state': 'running', 'role': 'replica'})
class TestCacheBuilder(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())
def setUp(self):
self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0',
'loop_wait': 10, 'retry_timeout': 10, 'labels': {'f': 'b'}})
class TestCacheBuilder(BaseTestKubernetes):
@patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map)
@patch('patroni.dcs.kubernetes.ObjectCache._watch')