From 0c516de147e152d2e5d1183a1d4c053cfc38e2d6 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 15 Feb 2019 13:35:04 +0100 Subject: [PATCH] Create headless service associated with $SCOPE-config endpoint (#958) if there is no service defined k8s assumes that endpoint is orphaned and removes it. Patroni tries to create the service only in case if use_endpoints is enabled if the following cases: 1. Upon start 2. When it tries to (re-)create the config endpoint If for some reason creation of the service has failed, Patroni will retry it on every cycle of HA loop. Usually it fails due to lack of permissions and if you don't want to give such permissions to the service account used by Patroni, you can create the service explicitly in the deployment manifest. --- kubernetes/patroni_k8s.yaml | 22 ++++++++++++++++++++++ patroni/dcs/kubernetes.py | 37 ++++++++++++++++++++++++++++--------- tests/test_kubernetes.py | 14 +++++++++++++- 3 files changed, 63 insertions(+), 10 deletions(-) diff --git a/kubernetes/patroni_k8s.yaml b/kubernetes/patroni_k8s.yaml index c1a8c7b0..4d45034a 100644 --- a/kubernetes/patroni_k8s.yaml +++ b/kubernetes/patroni_k8s.yaml @@ -1,3 +1,15 @@ +# headless service to avoid deletion of patronidemo-config endpoint +apiVersion: v1 +kind: Service +metadata: + name: patronidemo-config + labels: + application: patroni + cluster-name: patronidemo +spec: + clusterIP: None + +--- apiVersion: apps/v1beta1 kind: StatefulSet metadata: @@ -173,6 +185,16 @@ rules: - patch - update - watch +# The following privilege is only necessary for creation of headless service +# for patronidemo-config endpoint, in order to prevent cleaning it up by the +# k8s master. You can avoid giving this privilege by explicitly creating the +# service like it is done in this manifest (lines 2..10) +- apiGroups: + - "" + resources: + - services + verbs: + - create --- apiVersion: rbac.authorization.k8s.io/v1 diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 951f7aa0..525bd530 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -92,13 +92,14 @@ class Kubernetes(AbstractDCS): 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=config['pod_ip'])] + 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.set_retry_timeout(config['retry_timeout']) self.set_ttl(config.get('ttl') or 30) @@ -274,6 +275,24 @@ class Kubernetes(AbstractDCS): body = k8s_client.V1ConfigMap(metadata=metadata) return self.retry(func, self._namespace, body) if retry else func(self._namespace, body) + 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: + self._should_create_config_service = True + self._create_config_service() + return self.patch_or_create(self.config_path, annotations, resource_version, patch, retry) + + def _create_config_service(self): + metadata = k8s_client.V1ObjectMeta(namespace=self._namespace, name=self.config_path, labels=self._labels) + body = k8s_client.V1Service(metadata=metadata, spec=k8s_client.V1ServiceSpec(cluster_ip='None')) + try: + if not self._api.create_namespaced_service(self._namespace, body): + return + except Exception as e: + if not isinstance(e, k8s_client.rest.ApiException) or e.status != 409: # Service already exists + return logger.exception('create_config_service failed') + self._should_create_config_service = False + def _write_leader_optime(self, last_operation): """Unused""" @@ -288,9 +307,7 @@ class Kubernetes(AbstractDCS): if last_operation: annotations[self._OPTIME] = last_operation - subsets = self.__subsets - if subsets is not None and access_is_restricted: - subsets = [] + subsets = [] if access_is_restricted else self.__subsets ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, subsets=subsets) if ret: @@ -333,13 +350,13 @@ class Kubernetes(AbstractDCS): def set_config_value(self, value, index=None): patch = bool(index or self.cluster and self.cluster.config and self.cluster.config.index) - return self.patch_or_create(self.config_path, {self._CONFIG: value}, index, patch, False) + return self.patch_or_create_config({self._CONFIG: value}, index, patch, False) @catch_kubernetes_errors def touch_member(self, data, ttl=None, permanent=False): cluster = self.cluster if cluster and cluster.leader and cluster.leader.name == self._name: - role = 'master' + role = 'promoted' if data['role'] in ('replica', 'promoted') else 'master' elif data['state'] == 'running' and data['role'] != 'master': role = data['role'] else: @@ -354,12 +371,14 @@ 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: + self._create_config_service() return ret def initialize(self, create_new=True, sysid=""): cluster = self.cluster resource_version = cluster.config.index if cluster and cluster.config and cluster.config.index else None - return self.patch_or_create(self.config_path, {self._INITIALIZE: sysid}, resource_version) + return self.patch_or_create_config({self._INITIALIZE: sysid}, resource_version) def delete_leader(self): if self.cluster and isinstance(self.cluster.leader, Leader) and self.cluster.leader.name == self._name: @@ -367,7 +386,7 @@ class Kubernetes(AbstractDCS): self.reset_cluster() def cancel_initialization(self): - self.patch_or_create(self.config_path, {self._INITIALIZE: None}, self.cluster.config.index, True) + self.patch_or_create_config({self._INITIALIZE: None}, self.cluster.config.index, True) @catch_kubernetes_errors def delete_cluster(self): @@ -375,7 +394,7 @@ class Kubernetes(AbstractDCS): def set_history_value(self, value): patch = bool(self.cluster and self.cluster.config and self.cluster.config.index) - return self.patch_or_create(self.config_path, {self._HISTORY: value}, None, patch, False) + return self.patch_or_create_config({self._HISTORY: value}, None, patch, False) def set_sync_state_value(self, value, index=None): """Unused""" diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 03fa36f1..26c67758 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -72,7 +72,7 @@ class TestKubernetes(unittest.TestCase): @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', Mock(return_value=True)) def test_touch_member(self): - self.k.touch_member({}) + self.k.touch_member({'role': 'replica'}) self.k._name = 'p-1' self.k.touch_member({'state': 'running', 'role': 'replica'}) self.k.touch_member({'state': 'stopped', 'role': 'master'}) @@ -112,3 +112,15 @@ class TestKubernetes(unittest.TestCase): def test_set_history_value(self): self.k.set_history_value('{}') + + @patch('kubernetes.config.load_kube_config', Mock()) + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', Mock(return_value=True)) + @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', Mock()) + @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', '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'})