Make Patroni Kubernetes native (#500)

* Use ConfigMaps or Endpoins for leader elections and to keep cluster state
* Label pods with a postgres role
* change behavior of pip install. From now on it will not install all dependencies, you have to specify explicitly DCS you want to use Patroni with: `pip install patroni[etcd,zookeeper,kubernetes]`

.travis.yml

@@ -1,4 +1,4 @@
-sudo: false
+sudo: true
 dist: trusty
 language: python
 python:
@@ -7,11 +7,13 @@ env:
   global:
   - ETCDVERSION=3.0.17 ZKVERSION=3.4.9 CONSULVERSION=0.7.4
   - PYVERSIONS="2.7 3.4 3.5"
+  - BOTO_CONFIG=/doesnotexist
   matrix:
   - TEST_SUITE="python setup.py"
   - DCS="etcd" TEST_SUITE="behave"
   - DCS="exhibitor" TEST_SUITE="behave"
   - DCS="consul" TEST_SUITE="behave"
+  - DCS="kubernetes" TEST_SUITE="behave"
 cache:
   directories:
   - $HOME/mycache
@@ -51,6 +53,26 @@ install:
             ln -s $EC etcd
         }
 
+        function get_kubernetes() {
+            wget -O localkube "https://storage.googleapis.com/minikube/k8sReleases/v1.7.0/localkube-linux-amd64"
+            chmod +x localkube
+            sudo nohup ./localkube --logtostderr=true --enable-dns=false > localkube.log 2>&1 &
+
+            echo "Waiting for localkube to start..."
+            if ! timeout 120 sh -c "while ! curl -ks http://127.0.0.1:8080/ >/dev/null; do sleep 1; done"; then
+                sudo cat localkube.log
+                echo "localkube did not start"
+                exit 1
+            fi
+            echo "Check certificate permissions"
+            sudo chmod 644 /var/lib/localkube/certs/*
+            sudo ls -altr /var/lib/localkube/certs/
+
+            echo "Set up .kube/config"
+            mkdir ~/.kube
+            echo -e "apiVersion: v1\nclusters:\n- cluster:\n    certificate-authority: /var/lib/localkube/certs/ca.crt\n    server: https://127.0.0.1:8443\n  name: local\ncontexts:\n- context:\n    cluster: local\n    user: myself\n  name: local\ncurrent-context: local\nkind: Config\npreferences: {}\nusers:\n- name: myself\n  user:\n    client-certificate: /var/lib/localkube/certs/apiserver.crt\n    client-key: /var/lib/localkube/certs/apiserver.key\n" > ~/.kube/config
+        }
+
         function get_exhibitor() {
             ZC=~/mycache/zookeeper-${ZKVERSION}
             if [[ ! -d $ZC ]]; then
@@ -65,7 +87,6 @@ install:
                 echo -e 'HTTP/1.0 200 OK\nContent-Type: application/json\n\n{"servers":["127.0.0.1"],"port":2181}' \
                     | nc -l 8181 &> /dev/null
             done&
-            ZK_PID=$!
         }
 
         attempt_num=1
@@ -115,4 +136,5 @@ after_success:
   - fpv=$(basename $(readlink $HOME/virtualenv/python3.5)) && mv $HOME/mycache/${fpv} $HOME/virtualenv/${fpv}
   - coveralls
   - if [[ $TEST_SUITE != "behave" ]]; then python-codacy-coverage -r coverage.xml; fi
-  - if [[ $DCS == "exhibitor" ]]; then ~/mycache/zookeeper-${ZKVERSION}/bin/zkServer.sh stop; kill -9 $ZK_PID; fi
+  - if [[ $DCS == "exhibitor" ]]; then ~/mycache/zookeeper-${ZKVERSION}/bin/zkServer.sh stop; fi
+  - sudo kill $(jobs -p)

features/basic_replication.feature

@@ -21,9 +21,14 @@ Feature: basic replication
     And "sync" key in DCS has sync_standby=postgres2 after 10 seconds
 
   Scenario: check the basic failover in synchronous mode
-    When I kill postgres0
+    Given I run patronictl.py pause batman
-    Then postgres2 role is the primary after 22 seconds
+    Then I receive a response returncode 0
-    When I issue a PATCH request to http://127.0.0.1:8009/config with {"synchronous_mode": null, "master_start_timeout": 0}
+    When I sleep for 2 seconds
+    And I shut down postgres0
+    And I run patronictl.py resume batman 
+    Then I receive a response returncode 0
+    And postgres2 role is the primary after 24 seconds
+    When I issue a PATCH request to http://127.0.0.1:8010/config with {"synchronous_mode": null, "master_start_timeout": 0}
     Then I receive a response code 200
     When I add the table bar to postgres2
     Then table bar is present on postgres1 after 20 seconds

features/environment.py

@@ -7,6 +7,7 @@ import kazoo.exceptions
 import os
 import psutil
 import psycopg2
+import json
 import shutil
 import signal
 import six
@@ -98,6 +99,7 @@ class PatroniController(AbstractController):
         else:
             self.watchdog = None
 
+        self._scope = (custom_config or {}).get('scope', 'batman')
         self._config = self._make_patroni_test_config(name, custom_config)
         self._closables = []
 
@@ -125,6 +127,9 @@ class PatroniController(AbstractController):
     def _start(self):
         if self.watchdog:
             self.watchdog.start()
+        if isinstance(self._context.dcs_ctl, KubernetesController):
+            self._context.dcs_ctl.create_pod(self._name[8:], self._scope)
+            os.environ['PATRONI_KUBERNETES_POD_IP'] = '10.0.0.' + self._name[-1]
         return subprocess.Popen(['coverage', 'run', '--source=patroni', '-p', 'patroni.py', self._config],
                                 stdout=self._log, stderr=subprocess.STDOUT, cwd=self._work_directory)
 
@@ -132,6 +137,8 @@ class PatroniController(AbstractController):
         if postgres:
             return subprocess.call(['pg_ctl', '-D', self._data_dir, 'stop', '-mi', '-w'])
         super(PatroniController, self).stop(kill, timeout)
+        if isinstance(self._context.dcs_ctl, KubernetesController):
+            self._context.dcs_ctl.delete_pod(self._name[8:])
         if self.watchdog:
             self.watchdog.stop()
 
@@ -147,7 +154,7 @@ class PatroniController(AbstractController):
 
         with open(patroni_config_name) as f:
             config = yaml.safe_load(f)
-            config.pop('etcd')
+            config.pop('etcd', None)
 
         host = config['postgresql']['listen'].split(':')[0]
 
@@ -335,10 +342,6 @@ class AbstractDcsController(AbstractController):
     def query(self, key, scope='batman'):
         """ query for a value of a given key """
 
-    @abc.abstractmethod
-    def set(self, key, value):
-        """ set a value to a given key """
-
     @abc.abstractmethod
     def cleanup_service_tree(self):
         """ clean all contents stored in the tree used for the tests """
@@ -388,9 +391,6 @@ class ConsulController(AbstractDcsController):
         _, value = self._client.kv.get(self.path(key, scope))
         return value and value['Value'].decode('utf-8')
 
-    def set(self, key, value):
-        self._client.kv.put(self.path(key), value)
-
     def cleanup_service_tree(self):
         self._client.kv.delete(self.path(scope=''), recurse=True)
 
@@ -417,9 +417,6 @@ class EtcdController(AbstractDcsController):
         except etcd.EtcdKeyNotFound:
             return None
 
-    def set(self, key, value):
-        self._client.set(self.path(key), value)
-
     def cleanup_service_tree(self):
         try:
             self._client.delete(self.path(scope=''), recursive=True)
@@ -436,6 +433,76 @@ class EtcdController(AbstractDcsController):
             return False
 
 
+class KubernetesController(AbstractDcsController):
+
+    def __init__(self, context):
+        super(KubernetesController, self).__init__(context)
+        self._namespace = 'default'
+        self._labels = {"application": "patroni"}
+        self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items())
+        os.environ['PATRONI_KUBERNETES_LABELS'] = json.dumps(self._labels)
+        os.environ['PATRONI_KUBERNETES_USE_ENDPOINTS'] = 'true'
+
+        from kubernetes import client as k8s_client, config as k8s_config
+        k8s_config.load_kube_config(context='local')
+        self._client = k8s_client
+        self._api = self._client.CoreV1Api()
+
+    def _start(self):
+        pass
+
+    def create_pod(self, name, scope):
+        labels = self._labels.copy()
+        labels['cluster-name'] = scope
+        metadata = self._client.V1ObjectMeta(namespace=self._namespace, name=name, labels=labels)
+        spec = self._client.V1PodSpec(containers=[self._client.V1Container(name=name, image='empty')])
+        body = self._client.V1Pod(metadata=metadata, spec=spec)
+        self._api.create_namespaced_pod(self._namespace, body)
+
+    def delete_pod(self, name):
+        try:
+            self._api.delete_namespaced_pod(name, self._namespace, self._client.V1DeleteOptions())
+        except:
+            pass
+        while True:
+            try:
+                self._api.read_namespaced_pod(name, self._namespace)
+            except:
+                break
+
+    def query(self, key, scope='batman'):
+        if key.startswith('members/'):
+            pod = self._api.read_namespaced_pod(key[8:], self._namespace)
+            return (pod.metadata.annotations or {}).get('status', '')
+        else:
+            try:
+                e = self._api.read_namespaced_endpoints(scope + ('' if key == 'leader' else '-' + key), self._namespace)
+                if key == 'leader':
+                    return e.metadata.annotations[key]
+                else:
+                    return json.dumps(e.metadata.annotations)
+            except:
+                return None
+
+    def cleanup_service_tree(self):
+        try:
+            self._api.delete_collection_namespaced_pod(self._namespace, label_selector=self._label_selector)
+        except:
+            pass
+        try:
+            self._api.delete_collection_namespaced_endpoints(self._namespace, label_selector=self._label_selector)
+        except:
+            pass
+
+        while True:
+            result = self._api.list_namespaced_pod(self._namespace, label_selector=self._label_selector)
+            if len(result.items) < 1:
+                break
+
+    def _is_running(self):
+        return True
+
+
 class ZooKeeperController(AbstractDcsController):
 
     """ handles all zookeeper related tasks, used for the tests setup and cleanup """
@@ -455,9 +522,6 @@ class ZooKeeperController(AbstractDcsController):
         except kazoo.exceptions.NoNodeError:
             return None
 
-    def set(self, key, value):
-        self._client.set(self.path(key), value.encode('utf-8'))
-
     def cleanup_service_tree(self):
         try:
             self._client.delete(self.path(scope=''), recursive=True)
@@ -713,6 +777,7 @@ class WatchdogMonitor(object):
 
 # actions to execute on start/stop of the tests and before running invidual features
 def before_all(context):
+    os.environ.update({'PATRONI_RESTAPI_USERNAME': 'username', 'PATRONI_RESTAPI_PASSWORD': 'password'})
     context.ci = 'TRAVIS_BUILD_NUMBER' in os.environ or 'BUILD_NUMBER' in os.environ
     context.timeout_multiplier = 2 if context.ci else 1
     context.pctl = PatroniPoolController(context)

features/steps/patroni_api.py

@@ -1,3 +1,4 @@
+import base64
 import json
 import os
 import parse
@@ -78,11 +79,13 @@ def do_post_empty(context, url):
 @step('I issue a {request_method:w} request to {url:url} with {data}')
 def do_request(context, request_method, url, data):
     data = data and json.loads(data) or {}
+    headers = {'Authorization': 'Basic ' + base64.b64encode('username:password'.encode('utf-8')).decode('utf-8'),
+               'Content-Type': 'application/json'}
     try:
         if request_method == 'PATCH':
-            r = requests.patch(url, json=data)
+            r = requests.patch(url, headers=headers, json=data)
         else:
-            r = requests.post(url, json=data)
+            r = requests.post(url, headers=headers, json=data)
     except requests.exceptions.RequestException:
         context.status_code = None
         context.response = None

kubernetes/Dockerfile

@@ -0,0 +1,32 @@
+FROM postgres:9.6
+MAINTAINER Alexander Kukushkin <[email protected]>
+
+RUN export DEBIAN_FRONTEND=noninteractive \
+    && echo 'APT::Install-Recommends "0";\nAPT::Install-Suggests "0";' > /etc/apt/apt.conf.d/01norecommend \
+    && apt-get update -y \
+    && apt-get upgrade -y \
+    && apt-get install -y git curl jq python-psycopg2 python-yaml python-requests python-six python-pysocks \
+        python-dateutil python-pip python-prettytable python-wheel python-psutil python locales \
+
+    ## Make sure we have a en_US.UTF-8 locale available
+    && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \
+
+    && pip install setuptools pip --upgrade \
+    && pip install 'git+https://github.com/zalando/patroni.git@feature/k8s#egg=patroni[kubernetes]' \
+
+    && mkdir -p /home/postgres \
+    && chown postgres:postgres /home/postgres \
+
+    # Clean up
+    && apt-get remove -y git python-pip python-setuptools \
+    && apt-get autoremove -y \
+    && apt-get clean -y \
+    && rm -rf /var/lib/apt/lists/* /root/.cache
+
+ADD entrypoint.sh callback.py /
+
+EXPOSE 5432 8008
+ENV LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8
+USER postgres
+WORKDIR /home/postgres
+CMD ["/bin/bash", "/entrypoint.sh"]

kubernetes/callback.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python
+
+import logging
+import os
+import socket
+import sys
+import time
+
+from kubernetes import client as k8s_client, config as k8s_config
+from urllib3.exceptions import HTTPError
+from six.moves.http_client import HTTPException
+
+logger = logging.getLogger(__name__)
+
+
+class CoreV1Api(k8s_client.CoreV1Api):
+
+    def retry(func):
+        def wrapped(*args, **kwargs):
+            count = 0
+            while True:
+                try:
+                    return func(*args, **kwargs)
+                except (HTTPException, HTTPError, socket.error, socket.timeout):
+                    if count >= 10:
+                        raise
+                    logger.info('Throttling API requests...')
+                    time.sleep(2 ** count * 0.5)
+                    count += 1
+        return wrapped
+
+    @retry
+    def patch_namespaced_endpoints(self, *args, **kwargs):
+        return super(CoreV1Api, self).patch_namespaced_endpoints(*args, **kwargs)
+
+
+def patch_master_endpoint(api, namespace, cluster):
+    addresses = [k8s_client.V1EndpointAddress(ip=os.environ['POD_IP'])]
+    ports = [k8s_client.V1EndpointPort(port=5432)]
+    subsets = [k8s_client.V1EndpointSubset(addresses=addresses, ports=ports)]
+    body = k8s_client.V1Endpoints(subsets=subsets)
+    return api.patch_namespaced_endpoints(cluster, namespace, body)
+
+
+def main():
+    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+    if len(sys.argv) != 4 or sys.argv[1] not in ('on_start', 'on_stop', 'on_role_change'):
+        sys.exit('Usage: %s <action> <role> <cluster_name>', sys.argv[0])
+
+    action, role, cluster = sys.argv[1:4]
+
+    k8s_config.load_incluster_config()
+    k8s_api = CoreV1Api()
+
+    namespace = os.environ['KUBERNETES_NAMESPACE']
+
+    if role == 'master' and action in ('on_start', 'on_role_change'):
+        patch_master_endpoint(k8s_api, namespace, cluster)
+
+
+if __name__ == '__main__':
+    main()

kubernetes/entrypoint.sh

@@ -0,0 +1,36 @@
+#!/bin/bash
+
+cat > /home/postgres/patroni.yml <<__EOF__
+bootstrap:
+  dcs:
+    postgresql:
+      use_pg_rewind: true
+  initdb:
+  - auth-host: md5
+  - auth-local: trust
+  - encoding: UTF8
+  - locale: en_US.UTF-8
+  - data-checksums
+  pg_hba:
+  - host all all 0.0.0.0/0 md5
+  - host replication ${PATRONI_REPLICATION_USERNAME} ${POD_IP}/16    md5
+restapi:
+  connect_address: '${POD_IP}:8008'
+postgresql:
+  connect_address: '${POD_IP}:5432'
+  authentication:
+    superuser:
+      password: '${PATRONI_SUPERUSER_PASSWORD}'
+    replication:
+      password: '${PATRONI_REPLICATION_PASSWORD}'
+  callbacks:
+    on_start: /callback.py
+    on_stop: /callback.py
+    on_role_change: /callback.py
+__EOF__
+
+unset PATRONI_SUPERUSER_PASSWORD PATRONI_REPLICATION_PASSWORD
+export KUBERNETES_NAMESPACE=$PATRONI_KUBERNETES_NAMESPACE
+export POD_NAME=$PATRONI_NAME
+
+exec /usr/bin/python /usr/local/bin/patroni /home/postgres/patroni.yml

kubernetes/patroni_k8s.yaml

@@ -0,0 +1,122 @@
+apiVersion: apps/v1beta1
+kind: StatefulSet
+metadata:
+  name: &cluster_name patronidemo
+  labels:
+    application: patroni
+    cluster-name: *cluster_name
+spec:
+  replicas: 3
+  serviceName: *cluster_name
+  template:
+    metadata:
+      labels:
+        application: patroni
+        cluster-name: *cluster_name
+    spec:
+      containers:
+      - name: *cluster_name
+        image: patroni  # docker build -t patroni .
+        imagePullPolicy: IfNotPresent
+        ports:
+        - containerPort: 8008
+          protocol: TCP
+        - containerPort: 5432
+          protocol: TCP
+        volumeMounts:
+        - mountPath: /home/postgres/pgdata
+          name: pgdata
+        env:
+        - name: POD_IP
+          valueFrom:
+            fieldRef:
+              fieldPath: status.podIP
+        - name: PATRONI_KUBERNETES_NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        - name: PATRONI_KUBERNETES_LABELS
+          value: '{application: patroni, cluster-name: patronidemo}'
+        - name: PATRONI_SUPERUSER_USERNAME
+          value: postgres
+        - name: PATRONI_SUPERUSER_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: *cluster_name
+              key: superuser-password
+        - name: PATRONI_REPLICATION_USERNAME
+          value: standby
+        - name: PATRONI_REPLICATION_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: *cluster_name
+              key: replication-password
+        - name: PATRONI_SCOPE
+          value: *cluster_name
+        - name: PATRONI_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.name
+        - name: PATRONI_POSTGRESQL_DATA_DIR
+          value: /home/postgres/pgdata/pgroot/data
+        - name: PATRONI_POSTGRESQL_PGPASS
+          value: /tmp/pgpass
+        - name: PATRONI_POSTGRESQL_LISTEN
+          value: '0.0.0.0:5432'
+        - name: PATRONI_RESTAPI_LISTEN
+          value: '0.0.0.0:8008'
+      terminationGracePeriodSeconds: 0
+      volumes:
+      - name: pgdata
+        emptyDir: {}
+#  volumeClaimTemplates:
+#  - metadata:
+#      labels:
+#        application: spilo
+#        spilo-cluster: *cluster_name
+#      annotations:
+#        volume.alpha.kubernetes.io/storage-class: anything
+#      name: pgdata
+#    spec:
+#      accessModes:
+#      - ReadWriteOnce
+#      resources:
+#        requests:
+#          storage: 5Gi
+
+---
+apiVersion: v1
+kind: Endpoints
+metadata:
+  name: &cluster_name patronidemo
+  labels:
+    application: patroni
+    cluster-name: *cluster_name
+subsets: []
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: &cluster_name patronidemo
+  labels:
+    application: patroni
+    cluster-name: *cluster_name
+spec:
+  type: ClusterIP
+  ports:
+  - port: 5432
+    targetPort: 5432
+
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: &cluster_name patronidemo
+  labels:
+    application: patroni
+    cluster-name: *cluster_name
+type: Opaque
+data:
+  superuser-password: emFsYW5kbw==
+  replication-password: cmVwLXBhc3M=

patroni/__init__.py

@@ -39,7 +39,7 @@ class Patroni(object):
         while True:
             try:
                 cluster = self.dcs.get_cluster()
-                if cluster and cluster.config:
+                if cluster and cluster.config and cluster.config.data:
                     if self.config.set_dynamic_configuration(cluster.config):
                         self.dcs.reload_config(self.config)
                         self.watchdog.reload_config(self.config)
@@ -113,8 +113,8 @@ class Patroni(object):
 
             logger.info(self.ha.run_cycle())
 
-            cluster = self.dcs.cluster
+            if self.dcs.cluster and self.dcs.cluster.config and self.dcs.cluster.config.data \
-            if cluster and cluster.config and self.config.set_dynamic_configuration(cluster.config):
+                    and self.config.set_dynamic_configuration(self.dcs.cluster.config):
                 self.reload_config()
 
             if self.postgresql.role != 'uninitialized':

patroni/config.py

@@ -239,16 +239,25 @@ class Config(object):
 
         for param in list(os.environ.keys()):
             if param.startswith(Config.PATRONI_ENV_PREFIX):
-                name, suffix = (param[8:].rsplit('_', 1) + [''])[:2]
+                name, suffix = (param[8:].split('_', 1) + [''])[:2]
                 if name and suffix:
                     # PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT|..)
-                    if suffix in ('HOST', 'HOSTS', 'PORT', 'SRV', 'URL', 'PROXY', 'CACERT', 'CERT', 'KEY',
+                    if suffix in ('HOST', 'HOSTS', 'PORT', 'SRV', 'URL', 'PROXY', 'CACERT', 'CERT',
-                                  'VERIFY', 'TOKEN', 'CHECKS', 'DC') and '_' not in name:
+                                  'KEY', 'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'NAMESPACE', 'CONTEXT',
+                                  'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS'):
                         value = os.environ.pop(param)
                         if suffix == 'PORT':
                             value = value and parse_int(value)
-                        elif suffix in ('HOSTS', 'CHECKS'):
+                        elif suffix in ('HOSTS', 'PORTS', 'CHECKS'):
                             value = value and _parse_list(value)
+                        elif suffix == 'LABELS':
+                            if not value.strip().startswith('{'):
+                                value = '{{{0}}}'.format(value)
+                            try:
+                                value = yaml.safe_load(value)
+                            except Exception:
+                                logger.exception('Exception when parsing dict %s', value)
+                                value = None
                         if value:
                             ret[name.lower()][suffix.lower()] = value
                     # PATRONI_<username>_PASSWORD=<password>, PATRONI_<username>_OPTIONS=<option1,option2,...>

patroni/ctl.py

@@ -406,7 +406,7 @@ def remove(obj, cluster_name, fmt):
     if message != confirm:
         raise PatroniCtlException('You did not exactly type "{0}"'.format(message))
 
-    if cluster.leader:
+    if cluster.leader and cluster.leader.name:
         confirm = click.prompt('This cluster currently is healthy. Please specify the master name to continue')
         if confirm != cluster.leader.name:
             raise PatroniCtlException('You did not specify the current master of the cluster')

patroni/dcs/__init__.py

@@ -183,34 +183,38 @@ class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')):
     """
     >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader"}'))
     True
+    >>> 'Failover' in str(Failover.from_node(1, {"leader": "cluster_leader"}))
+    True
     >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader", "member": "cluster_candidate"}'))
     True
     >>> Failover.from_node(1, 'null') is None
-    True
+    False
     >>> n = '{"leader": "cluster_leader", "member": "cluster_candidate", "scheduled_at": "2016-01-14T10:09:57.1394Z"}'
     >>> 'tzinfo=' in str(Failover.from_node(1, n))
     True
     >>> Failover.from_node(1, None) is None
-    True
+    False
     >>> Failover.from_node(1, '{}') is None
-    True
+    False
     >>> 'abc' in Failover.from_node(1, 'abc:def')
     True
     """
     @staticmethod
     def from_node(index, value):
-        if not value:
+        if isinstance(value, dict):
-            return None
+            data = value
-
+        elif value:
-        try:
+            try:
-            data = json.loads(value)
+                data = json.loads(value)
-            if not data:
+                if not isinstance(data, dict):
-                return None
+                    data = {}
-        except ValueError:
+            except ValueError:
-            t = [a.strip() for a in value.split(':')]
+                t = [a.strip() for a in value.split(':')]
-            leader = t[0]
+                leader = t[0]
-            candidate = t[1] if len(t) > 1 else None
+                candidate = t[1] if len(t) > 1 else None
-            return Failover(index, leader, candidate, None) if leader or candidate else None
+                return Failover(index, leader, candidate, None) if leader or candidate else None
+        else:
+            data = {}
 
         if data.get('scheduled_at'):
             data['scheduled_at'] = dateutil.parser.parse(data['scheduled_at'])
@@ -258,8 +262,12 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
         True
         >>> SyncState.from_node(1, '{"leader": "leader"}').leader == "leader"
         True
+        >>> SyncState.from_node(1, {"leader": "leader"}).leader == "leader"
+        True
         """
-        if value:
+        if isinstance(value, dict):
+            data = value
+        elif value:
             try:
                 data = json.loads(value)
                 if not isinstance(data, dict):
@@ -345,8 +353,7 @@ class AbstractDCS(object):
             i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc...
         """
         self._name = config['name']
-        self._namespace = '/{0}'.format(config.get('namespace', '/service/').strip('/'))
+        self._base_path = os.path.join('/', config.get('namespace', '/service/').strip('/'), config['scope'])
-        self._base_path = '/'.join([self._namespace, config['scope']])
         self._set_loop_wait(config.get('loop_wait', 10))
 
         self._ctl = bool(config.get('patronictl', False))
@@ -449,7 +456,7 @@ class AbstractDCS(object):
             self._last_leader_operation = last_operation
 
     @abc.abstractmethod
-    def update_leader(self):
+    def _update_leader(self):
         """Update leader key (or session) ttl
 
         :returns: `!True` if leader key (or session) has been updated successfully.
@@ -458,6 +465,18 @@ class AbstractDCS(object):
         You have to use CAS (Compare And Swap) operation in order to update leader key,
         for example for etcd `prevValue` parameter must be used."""
 
+    def update_leader(self, last_operation):
+        """Update leader key (or session) ttl and optime/leader
+
+        :param last_operation: absolute xlog location in bytes
+        :returns: `!True` if leader key (or session) has been updated successfully.
+            If not, `!False` must be returned and current instance would be demoted."""
+
+        ret = self._update_leader()
+        if ret and last_operation:
+            self.write_leader_optime(last_operation)
+        return ret
+
     @abc.abstractmethod
     def attempt_to_acquire_leader(self, permanent=False):
         """Attempt to acquire leader lock
@@ -483,7 +502,6 @@ class AbstractDCS(object):
 
         if scheduled_at:
             failover_value['scheduled_at'] = scheduled_at.isoformat()
-
         return self.set_failover_value(json.dumps(failover_value, separators=(',', ':')), index)
 
     @abc.abstractmethod
@@ -496,7 +514,7 @@ class AbstractDCS(object):
         This method should create or update key with the name = '/members/' + `~self._name`
         and value = data in a given DCS.
 
-        :param data: json serialized information about instance (including connection strings)
+        :param data: information about instance (including connection strings)
         :param ttl: ttl for member key, optional parameter. If it is None `~self.member_ttl will be used`
         :param permanent: if set to `!True`, the member key will never expire.
          Used in patronictl for the external master.
@@ -533,8 +551,14 @@ class AbstractDCS(object):
     def delete_cluster(self):
         """Delete cluster from DCS"""
 
+    @staticmethod
+    def sync_state(leader, sync_standby):
+        """Build sync_state dict"""
+        return {'leader': leader, 'sync_standby': sync_standby}
+
     def write_sync_state(self, leader, sync_standby, index=None):
-        return self.set_sync_state_value(json.dumps({'leader': leader, 'sync_standby': sync_standby}), index=index)
+        sync_value = self.sync_state(leader, sync_standby)
+        return self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), index)
 
     @abc.abstractmethod
     def set_sync_state_value(self, value, index=None):

patroni/dcs/consul.py

@@ -1,4 +1,5 @@
 from __future__ import absolute_import
+import json
 import logging
 import os
 import socket
@@ -9,7 +10,7 @@ import urllib3
 from consul import ConsulException, NotFound, base
 from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState
 from patroni.exceptions import DCSError
-from patroni.utils import parse_bool, Retry, RetryFailedError
+from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError
 from urllib3.exceptions import HTTPError
 from six.moves.urllib.parse import urlencode, urlparse
 from six.moves.http_client import HTTPException
@@ -141,7 +142,7 @@ class Consul(AbstractDCS):
                             retry_exceptions=(ConsulInternalError, HTTPException,
                                               HTTPError, socket.error, socket.timeout))
 
-        self._my_member_data = None
+        self._my_member_data = {}
         kwargs = {}
         if 'url' in config:
             r = urlparse(config['url'])
@@ -311,12 +312,12 @@ class Consul(AbstractDCS):
             except Exception:
                 return False
 
-        if not create_member and member and data == self._my_member_data:
+        if not create_member and member and deep_compare(data, self._my_member_data):
             return True
 
         try:
             args = {} if permanent else {'acquire': self._session}
-            self._client.kv.put(self.member_path, data, **args)
+            self._client.kv.put(self.member_path, json.dumps(data, separators=(',', ':')), **args)
             self._my_member_data = data
             return True
         except Exception:
@@ -352,7 +353,7 @@ class Consul(AbstractDCS):
         return self._client.kv.put(self.leader_optime_path, last_operation)
 
     @catch_consul_errors
-    def update_leader(self):
+    def _update_leader(self):
         if self._session:
             self.retry(self._client.session.renew, self._session)
             self._last_session_refresh = time.time()
@@ -379,11 +380,11 @@ class Consul(AbstractDCS):
 
     @catch_consul_errors
     def set_sync_state_value(self, value, index=None):
-        return self._client.kv.put(self.sync_path, value, cas=index)
+        return self.retry(self._client.kv.put, self.sync_path, value, cas=index)
 
     @catch_consul_errors
     def delete_sync_state(self, index=None):
-        return self._client.kv.delete(self.sync_path, cas=index)
+        return self.retry(self._client.kv.delete, self.sync_path, cas=index)
 
     def watch(self, leader_index, timeout):
         if self.__do_not_watch:

patroni/dcs/etcd.py

@@ -1,5 +1,6 @@
 from __future__ import absolute_import
 import etcd
+import json
 import logging
 import os
 import urllib3.util.connection
@@ -467,6 +468,7 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def touch_member(self, data, ttl=None, permanent=False):
+        data = json.dumps(data, separators=(',', ':'))
         return self.retry(self._client.set, self.member_path, data, None if permanent else ttl or self._ttl)
 
     @catch_etcd_errors
@@ -499,7 +501,7 @@ class Etcd(AbstractDCS):
         return self._client.set(self.leader_optime_path, last_operation)
 
     @catch_etcd_errors
-    def update_leader(self):
+    def _update_leader(self):
         return self.retry(self._client.test_and_set, self.leader_path, self._name, self._name, self._ttl)
 
     @catch_etcd_errors
@@ -520,7 +522,7 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def set_sync_state_value(self, value, index=None):
-        return self._client.write(self.sync_path, value, prevIndex=index or 0)
+        return self.retry(self._client.write, self.sync_path, value, prevIndex=index or 0)
 
     @catch_etcd_errors
     def delete_sync_state(self, index=None):

patroni/dcs/kubernetes.py

@@ -0,0 +1,396 @@
+from __future__ import absolute_import
+import datetime
+import functools
+import json
+import logging
+import socket
+import sys
+import time
+
+from kubernetes import client as k8s_client, config as k8s_config, watch as k8s_watch
+from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState
+from patroni.exceptions import DCSError
+from patroni.utils import deep_compare, tzutc, Retry, RetryFailedError
+from urllib3.exceptions import HTTPError
+from six.moves.http_client import HTTPException
+
+logger = logging.getLogger(__name__)
+
+
+class KubernetesError(DCSError):
+    pass
+
+
+class KubernetesRetriableException(k8s_client.rest.ApiException):
+
+    def __init__(self, orig):
+        super(KubernetesRetriableException, self).__init__(orig.status, orig.reason)
+        self.body = orig.body
+        self.headers = orig.headers
+
+
+class CoreV1Api(object):
+
+    def __init__(self, use_endpoints=False):
+        self._api = k8s_client.CoreV1Api()
+        self._request_timeout = None
+        self._use_endpoints = use_endpoints
+
+    def set_timeout(self, timeout):
+        self._request_timeout = (1, timeout / 3.0)
+
+    def __getattr__(self, func):
+        if func.endswith('_kind'):
+            func = func[:-4] + ('endpoints' if self._use_endpoints else 'config_map')
+
+        def wrapper(*args, **kwargs):
+            if '_request_timeout' not in kwargs:
+                kwargs['_request_timeout'] = self._request_timeout
+            try:
+                return getattr(self._api, func)(*args, **kwargs)
+            except k8s_client.rest.ApiException as e:
+                if e.status in (502, 503, 504):  # XXX
+                    raise KubernetesRetriableException(e)
+                raise
+        return wrapper
+
+
+class Kubernetes(AbstractDCS):
+
+    def __init__(self, config):
+        self._labels = config['labels']
+        self._labels[config.get('scope_label', 'cluster-name')] = config['scope']
+        self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items())
+        self._namespace = config.get('namespace') or 'default'
+        self._role_label = config.get('role_label', 'role')
+        config['namespace'] = ''
+        super(Kubernetes, self).__init__(config)
+        self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
+                            retry_exceptions=(KubernetesRetriableException, HTTPException,
+                                              HTTPError, socket.error, socket.timeout))
+        self._ttl = None
+        try:
+            k8s_config.load_incluster_config()
+        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=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._api = CoreV1Api(use_endpoints)
+        self.set_retry_timeout(config['retry_timeout'])
+        self.set_ttl(config.get('ttl') or 30)
+        self._leader_observed_record = {}
+        self._leader_observed_time = None
+        self._leader_resource_version = None
+        self._leader_observed_subsets = []
+        self.__do_not_watch = False
+
+    def retry(self, *args, **kwargs):
+        return self._retry.copy()(*args, **kwargs)
+
+    def catch_kubernetes_errors(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            try:
+                return func(*args, **kwargs)
+            except (RetryFailedError, k8s_client.rest.ApiException,
+                    HTTPException, HTTPError, socket.error, socket.timeout):
+                return False
+        return wrapper
+
+    def client_path(self, path):
+        return super(Kubernetes, self).client_path(path)[1:].replace('/', '-')
+
+    @property
+    def leader_path(self):
+        return self._base_path[1:] if self.__subsets else super(Kubernetes, self).leader_path
+
+    def set_ttl(self, ttl):
+        ttl = int(ttl)
+        self.__do_not_watch = self._ttl != ttl
+        self._ttl = ttl
+
+    def set_retry_timeout(self, retry_timeout):
+        self._retry.deadline = retry_timeout
+        self._api.set_timeout(retry_timeout)
+
+    @staticmethod
+    def member(pod):
+        annotations = pod.metadata.annotations or {}
+        member = Member.from_node(pod.metadata.resource_version, pod.metadata.name, None, annotations.get('status', ''))
+        member.data['pod_labels'] = pod.metadata.labels
+        return member
+
+    def _load_cluster(self):
+        try:
+            # get list of members
+            response = self.retry(self._api.list_namespaced_pod, self._namespace, label_selector=self._label_selector)
+            members = [self.member(pod) for pod in response.items]
+
+            response = self.retry(self._api.list_namespaced_kind, self._namespace, label_selector=self._label_selector)
+            nodes = {item.metadata.name: item for item in response.items}
+
+            config = nodes.get(self.config_path)
+            metadata = config and config.metadata
+            annotations = metadata and metadata.annotations or {}
+
+            # get initialize flag
+            initialize = annotations.get(self._INITIALIZE)
+
+            # get global dynamic configuration
+            config = ClusterConfig.from_node(metadata and metadata.resource_version,
+                                             annotations.get(self._CONFIG) or '{}')
+
+            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 else []
+            annotations = metadata and metadata.annotations or {}
+
+            # get last leader operation
+            last_leader_operation = annotations.get(self._OPTIME)
+            last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation)
+
+            # get leader
+            leader_record = {n: annotations.get(n) for n in (self._LEADER, 'acquireTime',
+                             'ttl', 'renewTime', 'transitions') if n in annotations}
+            if (leader_record or self._leader_observed_record) and leader_record != self._leader_observed_record:
+                self._leader_observed_record = leader_record
+                self._leader_observed_time = time.time()
+
+            leader = leader_record.get(self._LEADER)
+            try:
+                ttl = int(leader_record.get('ttl')) or self._ttl
+            except (TypeError, ValueError):
+                ttl = self._ttl
+
+            if not metadata or not self._leader_observed_time or self._leader_observed_time + ttl < time.time():
+                leader = None
+
+            if metadata:
+                member = Member(-1, leader, None, {})
+                member = ([m for m in members if m.name == leader] or [member])[0]
+                leader = Leader(response.metadata.resource_version, None, member)
+
+            # failover key
+            failover = nodes.get(self.failover_path)
+            metadata = failover and failover.metadata
+            failover = Failover.from_node(metadata and metadata.resource_version, metadata and metadata.annotations)
+
+            # get synchronization state
+            sync = nodes.get(self.sync_path)
+            metadata = sync and sync.metadata
+            sync = SyncState.from_node(metadata and metadata.resource_version,  metadata and metadata.annotations)
+
+            self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync)
+        except Exception:
+            logger.exception('get_cluster')
+            raise KubernetesError('Kubernetes API is not responding properly')
+
+    @staticmethod
+    def compare_ports(p1, p2):
+        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):
+        """
+        >>> Kubernetes.subsets_changed([], [])
+        False
+        >>> Kubernetes.subsets_changed([], [k8s_client.V1EndpointSubset()])
+        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)
+        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)
+        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)
+        True
+        >>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p2, p1])]
+        >>> Kubernetes.subsets_changed(s1, s2)
+        False
+        """
+        if len(last_observed_subsets) != len(subsets):
+            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):
+            return True
+        if len(subsets[0].ports) == 1:
+            return not Kubernetes.compare_ports(last_observed_subsets[0].ports[0], subsets[0].ports[0])
+        observed_ports = {p.name: p for p in last_observed_subsets[0].ports}
+        for p in subsets[0].ports:
+            if p.name not in observed_ports or not Kubernetes.compare_ports(p, observed_ports.pop(p.name)):
+                return True
+        return False
+
+    @catch_kubernetes_errors
+    def patch_or_create(self, name, annotations, resource_version=None, patch=False, retry=True, subsets=None):
+        metadata = {'namespace': self._namespace, 'name': name, 'labels': self._labels, 'annotations': annotations}
+        if patch or resource_version:
+            if resource_version is not None:
+                metadata['resource_version'] = resource_version
+            func = functools.partial(self._api.patch_namespaced_kind, name)
+        else:
+            func = functools.partial(self._api.create_namespaced_kind)
+            # skip annotations with null values
+            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:
+            endpoints = {'metadata': metadata}
+            if self.subsets_changed(self._leader_observed_subsets, subsets):
+                endpoints['subsets'] = subsets
+            body = k8s_client.V1Endpoints(**endpoints)
+        else:
+            body = k8s_client.V1ConfigMap(metadata=metadata)
+        return self.retry(func, self._namespace, body) if retry else func(self._namespace, body)
+
+    def _write_leader_optime(self, last_operation):
+        """Unused"""
+
+    def _update_leader(self):
+        """Unused"""
+
+    def update_leader(self, last_operation):
+        now = datetime.datetime.now(tzutc).isoformat()
+        annotations = {self._LEADER: self._name, 'ttl': str(self._ttl), 'renewTime': now,
+                       'acquireTime': self._leader_observed_record.get('acquireTime') or now,
+                       'transitions': self._leader_observed_record.get('transitions') or '0'}
+        if last_operation:
+            annotations[self._OPTIME] = last_operation
+
+        ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, subsets=self.__subsets)
+        if ret:
+            self._leader_resource_version = ret.metadata.resource_version
+        return ret
+
+    def attempt_to_acquire_leader(self, permanent=False):
+        now = datetime.datetime.now(tzutc).isoformat()
+        annotations = {self._LEADER: self._name, 'ttl': str(sys.maxsize if permanent else self._ttl),
+                       'renewTime': now, 'acquireTime': now, 'transitions': '0'}
+        if self._leader_observed_record:
+            try:
+                transitions = int(self._leader_observed_record.get('transitions'))
+            except (TypeError, ValueError):
+                transitions = 0
+
+            if self._leader_observed_record.get(self._LEADER) != self._name:
+                transitions += 1
+            else:
+                annotations['acquireTime'] = self._leader_observed_record.get('acquireTime') or now
+            annotations['transitions'] = str(transitions)
+        ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, subsets=self.__subsets)
+        if ret:
+            self._leader_resource_version = ret.metadata.resource_version
+        else:
+            logger.info('Could not take out TTL lock')
+        return ret
+
+    def take_leader(self):
+        return self.attempt_to_acquire_leader()
+
+    def set_failover_value(self, value, index=None):
+        """Unused"""
+
+    def manual_failover(self, leader, candidate, scheduled_at=None, index=None):
+        annotations = {'leader': leader or None, 'member': candidate or None, 'scheduled_at': scheduled_at}
+        patch = bool(self.cluster and isinstance(self.cluster.failover, Failover) and self.cluster.failover.index)
+        return self.patch_or_create(self.failover_path, annotations, index, bool(index or patch), False)
+
+    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)
+
+    @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'
+        elif data['state'] == 'running' and data['role'] != 'master':
+            role = data['role']
+        else:
+            role = None
+
+        member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
+        pod_labels = member and member.data.pop('pod_labels', None)
+        ret = pod_labels is not None and pod_labels.get(self._role_label) == role and deep_compare(data, member.data)
+
+        if not ret:
+            metadata = {'namespace': self._namespace, 'name': self._name, 'labels': {self._role_label: role},
+                        '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)
+        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)
+
+    def delete_leader(self):
+        if self.cluster and isinstance(self.cluster.leader, Leader) and self.cluster.leader.name == self._name:
+            self.patch_or_create(self.leader_path, {self._LEADER: None}, self._leader_resource_version, True, False, [])
+            self.reset_cluster()
+
+    def cancel_initialization(self):
+        self.patch_or_create(self.config_path, {self._INITIALIZE: None}, self.cluster.config.index, True)
+
+    @catch_kubernetes_errors
+    def delete_cluster(self):
+        self.retry(self._api.delete_collection_namespaced_kind, self._namespace, label_selector=self._label_selector)
+
+    def set_sync_state_value(self, value, index=None):
+        """Unused"""
+
+    def write_sync_state(self, leader, sync_standby, index=None):
+        return self.patch_or_create(self.sync_path, self.sync_state(leader, sync_standby), index, False)
+
+    def delete_sync_state(self, index=None):
+        return self.write_sync_state(None, None, index)
+
+    def watch(self, leader_index, timeout):
+        if self.__do_not_watch:
+            self.__do_not_watch = False
+            return True
+
+        if leader_index:
+            end_time = time.time() + timeout
+            w = k8s_watch.Watch()
+            while timeout >= 1:
+                try:
+                    for event in w.stream(self._api.list_namespaced_kind, self._namespace,
+                                          resource_version=leader_index, timeout_seconds=int(timeout + 0.5),
+                                          field_selector='metadata.name=' + self.leader_path,
+                                          _request_timeout=(1, timeout + 1)):
+                        return event['raw_object'].get('metadata', {}).get('resourceVersion') != leader_index
+                    return False
+                except KeyboardInterrupt:
+                    raise
+                except Exception:
+                    logging.exception('watch')
+
+                timeout = end_time - time.time()
+
+        try:
+            return super(Kubernetes, self).watch(None, timeout)
+        finally:
+            self.event.clear()

patroni/dcs/zookeeper.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import time
 
@@ -6,6 +7,7 @@ from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.handlers.threading import SequentialThreadingHandler
 from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState
 from patroni.exceptions import DCSError
+from patroni.utils import deep_compare
 
 logger = logging.getLogger(__name__)
 
@@ -57,7 +59,7 @@ class ZooKeeper(AbstractDCS):
                                    max_delay=1, max_tries=-1, sleep_func=time.sleep))
         self._client.add_listener(self.session_listener)
 
-        self._my_member_data = None
+        self._my_member_data = {}
         self._fetch_cluster = True
 
         self._orig_kazoo_connect = self._client._connection._connect
@@ -242,7 +244,7 @@ class ZooKeeper(AbstractDCS):
     def touch_member(self, data, ttl=None, permanent=False):
         cluster = self.cluster
         member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
-        data = data.encode('utf-8')
+        encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8')
         if member and self._client.client_id is not None and member.session != self._client.client_id[0]:
             try:
                 self._client.delete_async(self.member_path).get(timeout=1)
@@ -253,11 +255,12 @@ class ZooKeeper(AbstractDCS):
             member = None
 
         if member:
-            if data == self._my_member_data:
+            if deep_compare(data, self._my_member_data):
                 return True
         else:
             try:
-                self._client.create_async(self.member_path, data, makepath=True, ephemeral=not permanent).get(timeout=1)
+                self._client.create_async(self.member_path, encoded_data, makepath=True,
+                                          ephemeral=not permanent).get(timeout=1)
                 self._my_member_data = data
                 return True
             except Exception as e:
@@ -265,7 +268,7 @@ class ZooKeeper(AbstractDCS):
                     logger.exception('touch_member')
                     return False
         try:
-            self._client.set_async(self.member_path, data).get(timeout=1)
+            self._client.set_async(self.member_path, encoded_data).get(timeout=1)
             self._my_member_data = data
             return True
         except Exception:
@@ -291,7 +294,7 @@ class ZooKeeper(AbstractDCS):
             logger.exception('Failed to update %s', self.leader_optime_path)
         return False
 
-    def update_leader(self):
+    def _update_leader(self):
         return True
 
     def delete_leader(self):

patroni/ha.py

@@ -86,14 +86,15 @@ class Ha(object):
         return self.dcs.attempt_to_acquire_leader()
 
     def update_lock(self, write_leader_optime=False):
-        ret = self.dcs.update_leader()
+        last_operation = None
+        if write_leader_optime:
+            try:
+                last_operation = self.state_handler.last_operation()
+            except Exception:
+                logger.exception('Exception when called state_handler.last_operation()')
+        ret = self.dcs.update_leader(last_operation)
         if ret:
             self.watchdog.keepalive()
-            if write_leader_optime:
-                try:
-                    self.dcs.write_leader_optime(self.state_handler.last_operation())
-                except Exception:
-                    pass
         return ret
 
     def has_lock(self):
@@ -132,7 +133,7 @@ class Ha(object):
                 scheduled_restart_data['schedule'] = scheduled_restart_data['schedule'].isoformat()
                 data['scheduled_restart'] = scheduled_restart_data
 
-            return self.dcs.touch_member(json.dumps(data, separators=(',', ':')))
+            return self.dcs.touch_member(data)
 
     def clone(self, clone_member=None, msg='(without leader)'):
         if self.state_handler.clone(clone_member):

requirements.txt

@@ -1,10 +1,10 @@
-urllib3>=1.9
+urllib3>=1.19.1,!=1.21
 boto
-psycopg2>=2.6.1
+psycopg2>=2.5.4
 PyYAML
 requests
 six >= 1.7
-kazoo==2.2.1
+kazoo>=1.3.1
 python-etcd>=0.4.3,<0.5
 python-consul>=0.7.0
 click>=4.1
@@ -13,3 +13,4 @@ tzlocal
 python-dateutil
 psutil
 cdiff
+kubernetes==3.0.0

setup.py

@@ -115,13 +115,23 @@ def read(fname):
 
 def setup_package():
     # Assemble additional setup commands
-    cmdclass = {}
+    cmdclass = {'test': PyTest}
-    cmdclass['test'] = PyTest
 
     # Some helper variables
     version = os.getenv('GO_PIPELINE_LABEL', VERSION)
 
-    install_reqs = get_install_requirements('requirements.txt')
+    install_requires = []
+    extras_require = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'],
+                      'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['kubernetes']}
+
+    for r in get_install_requirements('requirements.txt'):
+        extra = False
+        for e, v in extras_require.items():
+            if r.startswith(v[0]):
+                extras_require[e] = [r]
+                extra = True
+        if not extra:
+            install_requires.append(r)
 
     command_options = {'test': {'test_suite': ('setup.py', 'tests')}}
     if JUNIT_XML:
@@ -145,10 +155,10 @@ def setup_package():
         test_suite='tests',
         packages=find_packages(exclude=['tests', 'tests.*']),
         package_data={MAIN_PACKAGE: ["*.json"]},
-        install_requires=install_reqs,
+        install_requires=install_requires,
-        setup_requires=['flake8'],
+        extras_require=extras_require,
         cmdclass=cmdclass,
-        tests_require=['mock>=2.0.0', 'pytest-cov', 'pytest'],
+        tests_require=['flake8', 'mock>=2.0.0', 'pytest-cov', 'pytest'],
         command_options=command_options,
         entry_points={'console_scripts': CONSOLE_SCRIPTS},
     )

tests/test_config.py

@@ -49,6 +49,9 @@ class TestConfig(unittest.TestCase):
             'PATRONI_ETCD_CERT': '/cert',
             'PATRONI_ETCD_KEY': '/key',
             'PATRONI_CONSUL_HOST': '127.0.0.1:8500',
+            'PATRONI_KUBERNETES_LABELS': 'a:b:c',
+            'PATRONI_KUBERNETES_SCOPE_LABEL': 'a',
+            'PATRONI_KUBERNETES_PORTS': '[{"name": "postgresql"}]',
             'PATRONI_ZOOKEEPER_HOSTS': "'host1:2181','host2:2181'",
             'PATRONI_EXHIBITOR_HOSTS': 'host1,host2',
             'PATRONI_EXHIBITOR_PORT': '8181',

tests/test_consul.py

@@ -112,11 +112,11 @@ class TestConsul(unittest.TestCase):
     @patch.object(consul.Consul.KV, 'put', Mock(side_effect=[True, ConsulException]))
     def test_touch_member(self):
         self.c.refresh_session = Mock(return_value=True)
-        self.c.touch_member('balbla')
+        self.c.touch_member({'balbla': 'blabla'})
-        self.c.touch_member('balbla')
+        self.c.touch_member({'balbla': 'blabla'})
-        self.c.touch_member('balbla')
+        self.c.touch_member({'balbla': 'blabla'})
         self.c.refresh_session = Mock(return_value=False)
-        self.c.touch_member('balbla')
+        self.c.touch_member({'balbla': 'blabla'})
 
     @patch.object(consul.Consul.KV, 'put', Mock(return_value=False))
     def test_take_leader(self):
@@ -138,7 +138,7 @@ class TestConsul(unittest.TestCase):
 
     @patch.object(consul.Consul.Session, 'renew', Mock())
     def test_update_leader(self):
-        self.c.update_leader()
+        self.c.update_leader(None)
 
     @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True))
     def test_delete_leader(self):

tests/test_etcd.py

@@ -288,7 +288,7 @@ class TestEtcd(unittest.TestCase):
         self.etcd.write_leader_optime('0')
 
     def test_update_leader(self):
-        self.assertTrue(self.etcd.update_leader())
+        self.assertTrue(self.etcd.update_leader(None))
 
     def test_initialize(self):
         self.assertFalse(self.etcd.initialize())

tests/test_kubernetes.py

@@ -0,0 +1,103 @@
+import unittest
+
+from mock import Mock, patch
+from patroni.dcs.kubernetes import Kubernetes, KubernetesError, k8s_client, k8s_watch
+
+
+def mock_list_namespaced_config_map(self, *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))]
+    metadata.update({'name': 'test-leader', 'annotations': {'optime': '1234', 'leader': 'p-0', 'ttl': '30s'}})
+    items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
+    metadata.update({'name': 'test-failover', 'annotations': {'leader': 'p-0'}})
+    items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
+    metadata.update({'name': 'test-sync', 'annotations': {'leader': 'p-0'}})
+    items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
+    metadata = k8s_client.V1ObjectMeta(resource_version='1')
+    return k8s_client.V1ConfigMapList(metadata=metadata, items=items)
+
+
+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)]
+    return k8s_client.V1PodList(items=items)
+
+
+@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', Mock())
+@patch.object(k8s_client.CoreV1Api, 'create_namespaced_config_map', Mock())
+class TestKubernetes(unittest.TestCase):
+
+    @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)
+    def setUp(self):
+        self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, 'labels': {'f': 'b'}})
+        with patch('time.time', Mock(return_value=1)):
+            self.k.get_cluster()
+
+    @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)
+    def test_get_cluster(self):
+        self.k.get_cluster()
+        with patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', 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', 'retry_timeout': 10,
+                        'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
+        self.assertIsNotNone(k.update_leader('123'))
+
+    def test_take_leader(self):
+        self.k.take_leader()
+        self.k._leader_observed_record['leader'] = 'test'
+        self.k.patch_or_create = Mock(return_value=False)
+        self.k.take_leader()
+
+    def test_manual_failover(self):
+        self.k.manual_failover('foo', 'bar')
+
+    def test_set_config_value(self):
+        self.k.set_config_value('{}')
+
+    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', Mock(return_value=True))
+    def test_touch_member(self):
+        self.k.touch_member({})
+        self.k._name = 'p-1'
+        self.k.touch_member({'state': 'running', 'role': 'replica'})
+        self.k.touch_member({'state': 'stopped', 'role': 'master'})
+
+    def test_initialize(self):
+        self.k.initialize()
+
+    def test_delete_leader(self):
+        self.k.delete_leader()
+
+    def test_cancel_initialization(self):
+        self.k.cancel_initialization()
+
+    @patch.object(k8s_client.CoreV1Api, 'delete_collection_namespaced_config_map',
+                  Mock(side_effect=k8s_client.rest.ApiException(500, '')))
+    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', '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)
+        self.k.watch(None, 0)
+        with patch.object(k8s_watch.Watch, 'stream',
+                          Mock(side_effect=[Exception, [], KeyboardInterrupt,
+                                            [{'raw_object': {'metadata': {'resourceVersion': '2'}}}]])):
+            self.assertFalse(self.k.watch('1', 2))
+            self.assertRaises(KeyboardInterrupt, self.k.watch, '1', 2)
+            self.assertTrue(self.k.watch('1', 2))

tests/test_zookeeper.py

@@ -58,11 +58,11 @@ class MockKazooClient(Mock):
             raise TypeError("Invalid type for 'path' (string expected)")
         if not isinstance(value, (six.binary_type,)):
             raise TypeError("Invalid type for 'value' (must be a byte string)")
-        if value == b'Exception':
+        if b'Exception' in value:
             raise Exception
         if path.endswith('/initialize') or path == '/service/test/optime/leader':
             raise Exception
-        elif value == b'retry' or (value == b'exists' and self.exists):
+        elif b'retry' in value or (b'exists' in value and self.exists):
             raise NodeExistsError
 
     def create_async(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False):
@@ -76,10 +76,10 @@ class MockKazooClient(Mock):
             raise TypeError("Invalid type for 'value' (must be a byte string)")
         if path == '/service/bla/optime/leader':
             raise Exception
-        if path == '/service/test/members/bar' and value == b'retry':
+        if path == '/service/test/members/bar' and b'retry' in value:
             return
         if path in ('/service/test/failover', '/service/test/config', '/service/test/sync'):
-            if value == b'Exception':
+            if b'Exception' in value:
                 raise Exception
             elif value == b'ok':
                 return
@@ -145,7 +145,7 @@ class TestZooKeeper(unittest.TestCase):
         self.assertRaises(ZooKeeperError, self.zk.get_cluster)
         cluster = self.zk.get_cluster()
         self.assertIsInstance(cluster.leader, Leader)
-        self.zk.touch_member('foo')
+        self.zk.touch_member({'foo': 'foo'})
 
     def test_delete_leader(self):
         self.assertTrue(self.zk.delete_leader())
@@ -169,17 +169,17 @@ class TestZooKeeper(unittest.TestCase):
     def test_touch_member(self):
         self.zk._name = 'buzz'
         self.zk.get_cluster()
-        self.zk.touch_member('new')
+        self.zk.touch_member({'new': 'new'})
         self.zk._name = 'bar'
-        self.zk.touch_member('new')
+        self.zk.touch_member({'new': 'new'})
         self.zk._name = 'na'
         self.zk._client.exists = 1
-        self.zk.touch_member('Exception')
+        self.zk.touch_member({'Exception': 'Exception'})
         self.zk._name = 'bar'
-        self.zk.touch_member('retry')
+        self.zk.touch_member({'retry': 'retry'})
         self.zk._fetch_cluster = True
         self.zk.get_cluster()
-        self.zk.touch_member('retry')
+        self.zk.touch_member({'retry': 'retry'})
 
     def test_take_leader(self):
         self.zk.take_leader()
@@ -187,7 +187,7 @@ class TestZooKeeper(unittest.TestCase):
             self.zk.take_leader()
 
     def test_update_leader(self):
-        self.assertTrue(self.zk.update_leader())
+        self.assertTrue(self.zk.update_leader(None))
 
     def test_write_leader_optime(self):
         self.zk.last_leader_operation = '0'