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

@@ -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

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"""

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