diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 3325bed0..85fbe212 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -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 = [] - 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.__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)}) + 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 diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index e82870f4..1a04ce9a 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -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() + 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')