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'})