Release 1.5.3 (#889)

* Bump version
* Update release notes

docs/SETTINGS.rst

@@ -74,6 +74,8 @@ Most of the parameters are optional, but you have to specify one of the **host**
 -  **key**: (optional) file with the client key. Can be empty if the key is part of **cert**.
 -  **dc**: (optional) Datacenter to communicate with. By default the datacenter of the host is used.
 -  **checks**: (optional) list of Consul health checks used for the session. If not specified Consul will use "serfHealth" in additional to the TTL based check created by Patroni. Additional checks, in particular the "serfHealth", may cause the leader lock to expire faster than in `ttl` seconds when the leader instance becomes unavailable
+-  **register\_service**: (optional) whether or not to register a service with the name defined by the scope parameter and the tag master, replica or standby-leader depending on the node's role. Defaults to **false**
+-  **service\_check\_interval**: (optional) how often to perform health check against registered url
 
 Etcd
 ----
@@ -149,8 +151,10 @@ PostgreSQL
 
 REST API
 -------- 
--  **connect\_address**: IP address (or hostname) and port, to access the Patroni's REST API. It can serve as a endpoint for HTTP health checks (read below about the "listen" REST API parameter), and also for user queries (either directly or via the REST API), as well as for the health checks done by the cluster members during leader elections (for example, to determine whether the master is still running, or if there is a node which has a WAL position that is ahead of the one doing the query; etc.) The connect_address is put in the member key in DCS, making it possible to translate the member name into the address to connect to its REST API.
--  **listen**: IP address (or hostname) and port that Patroni will listen to for the REST API - to provide also the same health checks and cluster messaging between the participating nodes, as described above. to provide health-check information for HAProxy (or any other load balancer capable of doing a HTTP "OPTION" or "GET" checks)  
+-  **connect\_address**: IP address (or hostname) and port, to access the Patroni's REST API. All the members of the cluster must be able to connect to this address, so unless the Patroni setup is intended for a demo inside the localhost, this address must be a non "localhost" or loopback addres (ie: "localhost" or "127.0.0.1"). It can serve as a endpoint for HTTP health checks (read below about the "listen" REST API parameter), and also for user queries (either directly or via the REST API), as well as for the health checks done by the cluster members during leader elections (for example, to determine whether the master is still running, or if there is a node which has a WAL position that is ahead of the one doing the query; etc.) The connect_address is put in the member key in DCS, making it possible to translate the member name into the address to connect to its REST API.
+
+-  **listen**: IP address (or hostname) and port that Patroni will listen to for the REST API - to provide also the same health checks and cluster messaging between the participating nodes, as described above. to provide health-check information for HAProxy (or any other load balancer capable of doing a HTTP "OPTION" or "GET" checks).
+
 -  **Optional**:
         -  **authentication**:
             -  **username**: Basic-auth username to protect unsafe REST API endpoints.

docs/releases.rst

@@ -3,6 +3,36 @@
 Release notes
 =============
 
+Version 1.5.3
+-------------
+
+Compatibility and bugfix release.
+
+- Improve stability when running with python3 against zookeeper (Alexander Kukushkin)
+
+  Change of `loop_wait` was causing Patroni to disconnect from zookeeper and never reconnect back.
+
+- Fix broken compatibility with postgres 9.3 (Alexander)
+
+  When opening a replication connection we should specify replication=1, beacuse 9.3 does not understand replication='database'
+
+- Make sure we refresh Consul session at least once per HA loop and improve handling of consul sessions exceptions (Alexander)
+
+  Restart of local consul agent invalidates all sessions related to the node. Not calling session refresh on time and not doing proper handling of session errors was causing demote of the primary.
+
+Version 1.5.2
+-------------
+
+Compatibility and bugfix release.
+
+- Compatibility with kazoo-2.6.0 (Alexander Kukushkin)
+
+  In order to make sure that requests are performed with an appropriate timeout, Patroni redefines create_connection method from python-kazoo module. The last release of kazoo slightly changed the way how create_connection method is called.
+
+- Fix Patroni crash when Consul cluster loses the leader (Alexander)
+
+  The crash was happening due to incorrect implementation of touch_member method, it should return boolean and not raise any exceptions.
+
 Version 1.5.1
 -------------
 

docs/replica_bootstrap.rst

@@ -101,7 +101,7 @@ example: pgbackrest
             - pgbackrest
             - basebackup
         pgbackrest:
-            command: /usr/bin/pgbackrest --stanza=mydb --deltarestore
+            command: /usr/bin/pgbackrest --stanza=<scope> --delta restore
             keep_data: True
             no_params: True
         basebackup:

extras/README.md

@@ -1,11 +1,11 @@
 ### confd
 
-`confd` directory contains haproxy template files for the [confd](https://github.com/kelseyhightower/confd) -- lightweight configuration management tool
+`confd` directory contains haproxy and pgbouncer template files for the [confd](https://github.com/kelseyhightower/confd) -- lightweight configuration management tool
 You need to copy content of `confd` directory into /etcd/confd and run confd service:
 ```bash
 $ confd -prefix=/service/$PATRONI_SCOPE -backend etcd -node $PATRONI_ETCD_URL -interval=10
 ```
-It will periodically update haproxy.cfg with the actual list of Patroni nodes from `etcd` and "reload" haproxy when it is necessary.
+It will periodically update haproxy.cfg and pgbouncer.ini with the actual list of Patroni nodes from `etcd` and "reload" haproxy and pgbouncer.ini when it is necessary.
 
 
 ### startup-scripts

extras/confd/conf.d/pgbouncer.toml

@@ -0,0 +1,12 @@
+[template]
+prefix = "/service/batman"
+owner = "postgres"
+mode = "0644"
+src = "pgbouncer.tmpl"
+dest = "/etc/pgbouncer/pgbouncer.ini"
+
+reload_cmd = "systemctl reload pgbouncer"
+
+keys = [
+    "/members/","/leader"
+]

extras/confd/templates/pgbouncer.tmpl

@@ -0,0 +1,17 @@
+[databases]
+{{with get "/leader"}}{{$leader := .Value}}{{$leadkey := printf "/members/%s" $leader}}{{with get $leadkey}}{{$data := json .Value}}{{$hostport := base (replace (index (split $data.conn_url "/") 2) "@" "/" -1)}}{{ $host := base (index (split $hostport ":") 0)}}{{ $port := base (index (split $hostport ":") 1)}}* = host={{ $host }} port={{ $port }} pool_size=10{{end}}{{end}}
+
+[pgbouncer]
+logfile = /var/log/postgresql/pgbouncer.log
+pidfile = /var/run/postgresql/pgbouncer.pid
+listen_addr = *
+listen_port = 6432
+unix_socket_dir = /var/run/postgresql
+auth_type = trust
+auth_file = /etc/pgbouncer/userlist.txt
+auth_hba_file = /etc/pgbouncer/pg_hba.txt
+admin_users = pgbouncer
+stats_users = pgbouncer
+pool_mode = session
+max_client_conn = 100
+default_pool_size = 20

features/environment.py

@@ -117,12 +117,24 @@ class PatroniController(AbstractController):
         except IOError:
             return None
 
-    def add_tag_to_config(self, tag, value):
+    @staticmethod
+    def recursive_update(dst, src):
+        for k, v in src.items():
+            if k in dst and isinstance(dst[k], dict):
+                PatroniController.recursive_update(dst[k], v)
+            else:
+                dst[k] = v
+
+    def update_config(self, custom_config):
         with open(self._config) as r:
             config = yaml.safe_load(r)
-            config['tags']['tag'] = value
+            self.recursive_update(config, custom_config)
             with open(self._config, 'w') as w:
                 yaml.safe_dump(config, w, default_flow_style=False)
+        self._scope = config.get('scope', 'batman')
+
+    def add_tag_to_config(self, tag, value):
+        self.update_config({'tags': {tag: value}})
 
     def _start(self):
         if self.watchdog:
@@ -174,13 +186,7 @@ class PatroniController(AbstractController):
                 config['bootstrap']['initdb'].extend([{'auth': 'md5'}, {'auth-host': 'md5'}])
 
         if custom_config is not None:
-            def recursive_update(dst, src):
-                for k, v in src.items():
-                    if k in dst and isinstance(dst[k], dict):
-                        recursive_update(dst[k], v)
-                    else:
-                        dst[k] = v
-            recursive_update(config, custom_config)
+            self.recursive_update(config, custom_config)
 
         if config['postgresql'].get('callbacks', {}).get('on_role_change'):
             config['postgresql']['callbacks']['on_role_change'] += ' ' + str(self.__PORT)

features/patroni_api.feature

@@ -43,7 +43,7 @@ Scenario: check dynamic configuration change via DCS
 	And I receive a response loop_wait 2
 	When I issue a GET request to http://127.0.0.1:8008/patroni
 	Then I receive a response code 200
-	And I receive a response tags {'tag': 'new_value'}
+	And I receive a response tags {'new_tag': 'new_value'}
 
 Scenario: check API requests for the primary-replica pair in the pause mode
 	Given I run patronictl.py pause batman

features/standby_cluster.feature

@@ -2,12 +2,16 @@ Feature: standby cluster
   Scenario: check permanent logical slots are preserved on failover/switchover
     Given I start postgres1
     Then postgres1 is a leader after 10 seconds
-    When I issue a PATCH request to http://127.0.0.1:8009/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
+    And I sleep for 2 seconds
+    When I issue a PATCH request to http://127.0.0.1:8009/config with {"loop_wait": 2, "slots": {"pm_1": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}}
     Then I receive a response code 200
-    When I issue a PATCH request to http://127.0.0.1:8009/config with {"slots": {"pm_1": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}}
+    And Response on GET http://127.0.0.1:8009/config contains slots after 10 seconds
+    And I sleep for 2 seconds
+    When I issue a PATCH request to http://127.0.0.1:8009/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
     Then I receive a response code 200
     When I start postgres0 with callback configured
     Then "members/postgres0" key in DCS has state=running after 10 seconds
+    And replication works from postgres1 to postgres0 after 15 seconds
     When I shut down postgres1
     Then postgres0 is a leader after 10 seconds
     And I sleep for 2 seconds
@@ -18,8 +22,7 @@ Feature: standby cluster
   Scenario: check replication of a single table in a standby cluster
     Given I start postgres1 in a standby cluster batman1 as a clone of postgres0
     Then postgres1 is a leader of batman1 after 10 seconds
-    When I issue a PATCH request to http://127.0.0.1:8009/config with {"ttl": 20, "loop_wait": 2}
-    And I add the table foo to postgres0
+    When I add the table foo to postgres0
     Then table foo is present on postgres1 after 20 seconds
     When I start postgres2 in a cluster batman1
     Then postgres2 role is the replica after 24 seconds

features/steps/standby_cluster.py

@@ -30,12 +30,10 @@ def start_patroni(context, name, cluster_name):
 
 @step('I start {name:w} in a standby cluster {cluster_name:w} as a clone of {name2:w}')
 def start_patroni_stanby_cluster(context, name, cluster_name, name2):
-    ctl = context.pctl._processes.pop(name, None)
     # we need to remove patroni.dynamic.json in order to "bootstrap" standby cluster with existing PGDATA
-    if ctl:
-        os.unlink(os.path.join(ctl._data_dir, 'patroni.dynamic.json'))
+    os.unlink(os.path.join(context.pctl._processes[name]._data_dir, 'patroni.dynamic.json'))
     port = context.pctl._processes[name2]._connkwargs.get('port')
-    return context.pctl.start(name, custom_config={
+    context.pctl._processes[name].update_config({
         "scope": cluster_name,
         "bootstrap": {
             "dcs": {
@@ -47,6 +45,7 @@ def start_patroni_stanby_cluster(context, name, cluster_name, name2):
             }
         }
     })
+    return context.pctl.start(name)
 
 
 @step('{pg_name1:w} is replicating from {pg_name2:w} after {timeout:d} seconds')

kubernetes/Dockerfile

@@ -1,32 +1,34 @@
-FROM postgres:9.6
+FROM postgres:11
 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 \
-
+    && apt-cache depends patroni | sed -n -e 's/.* Depends: \(python3-.\+\)$/\1/p' \
+            | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \
+            | xargs apt-get install -y vim-tiny curl jq locales git python3-pip python3-wheel \
     ## 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#egg=patroni[kubernetes]' \
-
-    && mkdir -p /home/postgres \
-    && chown postgres:postgres /home/postgres \
-
+    && pip3 install setuptools \
+    && pip3 install 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \
+    && PGHOME=/home/postgres \
+    && mkdir -p $PGHOME \
+    && chown postgres $PGHOME \
+    && sed -i "s|/var/lib/postgresql.*|$PGHOME:/bin/bash|" /etc/passwd \
+    # Set permissions for OpenShift
+    && chmod 775 $PGHOME \
+    && chmod 664 /etc/passwd \
     # Clean up
-    && apt-get remove -y git python-pip python-setuptools \
+    && apt-get remove -y git python3-pip python3-wheel \
     && apt-get autoremove -y \
     && apt-get clean -y \
     && rm -rf /var/lib/apt/lists/* /root/.cache
 
-ADD entrypoint.sh callback.py /
+ADD entrypoint.sh /
 
 EXPOSE 5432 8008
-ENV LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8
+ENV LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8 EDITOR=/usr/bin/editor
 USER postgres
 WORKDIR /home/postgres
 CMD ["/bin/bash", "/entrypoint.sh"]

kubernetes/callback.py

@@ -1,62 +0,0 @@
-#!/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

@@ -1,5 +1,12 @@
 #!/bin/bash
 
+if [[ $UID -ge 10000 ]]; then
+    GID=$(id -g)
+    sed -e "s/^postgres:x:[^:]*:[^:]*:/postgres:x:$UID:$GID:/" /etc/passwd > /tmp/passwd
+    cat /tmp/passwd > /etc/passwd
+    rm /tmp/passwd
+fi
+
 cat > /home/postgres/patroni.yml <<__EOF__
 bootstrap:
   dcs:
@@ -13,24 +20,20 @@ bootstrap:
   - data-checksums
   pg_hba:
   - host all all 0.0.0.0/0 md5
-  - host replication ${PATRONI_REPLICATION_USERNAME} ${POD_IP}/16    md5
+  - host replication ${PATRONI_REPLICATION_USERNAME} ${PATRONI_KUBERNETES_POD_IP}/16 md5
 restapi:
-  connect_address: '${POD_IP}:8008'
+  connect_address: '${PATRONI_KUBERNETES_POD_IP}:8008'
 postgresql:
-  connect_address: '${POD_IP}:5432'
+  connect_address: '${PATRONI_KUBERNETES_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
+exec /usr/bin/python3 /usr/local/bin/patroni /home/postgres/patroni.yml

kubernetes/openshift-example/README.md

@@ -0,0 +1,49 @@
+# Patroni OpenShift Configuration
+Patroni can be run in OpenShift. Based on the kubernetes configuration, the Dockerfile and Entrypoint has been modified to support the dynamic UID/GID configuration that is applied in OpenShift. This can be run under the standard `restricted` SCC. 
+
+# Examples
+
+## Create test project
+
+```
+oc new-project patroni-test
+```
+
+## Build the image
+
+Note: Update the references when merged upstream. 
+Note: If deploying as a template for multiple users, the following commands should be performed in a shared namespace like `openshift`. 
+
+```
+oc import-image postgres:10 --confirm -n openshift
+oc new-build https://github.com/zalando/patroni --context-dir=kubernetes -n openshift
+```
+
+## Deploy the Image 
+Two configuration templates exist in [templates](templates) directory: 
+- Patroni Ephemeral
+- Patroni Persistent
+
+The only difference is whether or not the statefulset requests persistent storage. 
+
+## Create the Template
+Install the template into the `openshift` namespace if this should be shared across projects: 
+
+```
+oc create -f templates/template_patroni_ephemeral.yml -n openshift
+```
+
+Then, from your own project: 
+
+```
+oc new-app patroni-pgsql-ephemeral
+```
+
+Once the pods are running, two configmaps should be available: 
+
+```
+$ oc get configmap
+NAME                DATA      AGE
+patroniocp-config   0         1m
+patroniocp-leader   0         1m
+```

kubernetes/openshift-example/templates/template_patroni_ephemeral.yml

@@ -0,0 +1,287 @@
+apiVersion: v1
+kind: Template
+metadata:
+  name: patroni-pgsql-ephemeral
+  annotations:
+    description: |-
+      Patroni Postgresql database cluster, without persistent storage.
+
+      WARNING: Any data stored will be lost upon pod destruction. Only use this template for testing.
+    iconClass: icon-postgresql
+    openshift.io/display-name: Patroni Postgresql (Ephemeral)
+    openshift.io/long-description: This template deploys a a patroni postgresql HA cluster without persistent storage.
+    tags: postgresql
+objects:
+- apiVersion: v1
+  kind: Service
+  metadata:
+    creationTimestamp: null
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${PATRONI_CLUSTER_NAME}
+  spec:
+    ports:
+    - port: 5432
+      protocol: TCP
+      targetPort: 5432
+    sessionAffinity: None
+    type: ClusterIP
+  status:
+    loadBalancer: {}
+- apiVersion: v1
+  kind: Service
+  metadata:
+    creationTimestamp: null
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${PATRONI_MASTER_SERVICE_NAME}
+  spec:
+    ports:
+    - port: 5432
+      protocol: TCP
+      targetPort: 5432
+    selector:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+      role: master
+    sessionAffinity: None
+    type: ClusterIP
+  status:
+    loadBalancer: {}
+- apiVersion: v1
+  kind: Secret
+  metadata:
+    name: ${PATRONI_CLUSTER_NAME}
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+  stringData:
+    superuser-password: ${PATRONI_SUPERUSER_PASSWORD}
+    replication-password: ${PATRONI_REPLICATION_PASSWORD}
+- apiVersion: v1
+  kind: Service
+  metadata:
+    creationTimestamp: null
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${PATRONI_REPLICA_SERVICE_NAME}
+  spec:
+    ports:
+    - port: 5432
+      protocol: TCP
+      targetPort: 5432
+    selector:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+      role: replica
+    sessionAffinity: None
+    type: ClusterIP
+  status:
+    loadBalancer: {}
+- apiVersion: apps/v1
+  kind: StatefulSet
+  metadata:
+    creationTimestamp: null
+    generation: 3
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${APPLICATION_NAME}
+  spec:
+    podManagementPolicy: OrderedReady
+    replicas: 3
+    revisionHistoryLimit: 10
+    selector:
+      matchLabels:
+        application: ${APPLICATION_NAME}
+        cluster-name: ${PATRONI_CLUSTER_NAME}
+    serviceName: ${APPLICATION_NAME}
+    template:
+      metadata:
+        creationTimestamp: null
+        labels:
+          application: ${APPLICATION_NAME}
+          cluster-name: ${PATRONI_CLUSTER_NAME}
+      spec:
+        containers:
+        - env:
+          - name: PATRONI_KUBERNETES_POD_IP
+            valueFrom:
+              fieldRef:
+                apiVersion: v1
+                fieldPath: status.podIP
+          - name: PATRONI_KUBERNETES_NAMESPACE
+            valueFrom:
+              fieldRef:
+                apiVersion: v1
+                fieldPath: metadata.namespace
+          - name: PATRONI_KUBERNETES_LABELS
+            value: '{application: ${APPLICATION_NAME}, cluster-name: ${PATRONI_CLUSTER_NAME}}'
+          - name: PATRONI_SUPERUSER_USERNAME
+            value: ${PATRONI_SUPERUSER_USERNAME}
+          - name: PATRONI_SUPERUSER_PASSWORD
+            valueFrom:
+              secretKeyRef:
+                key: superuser-password
+                name: ${PATRONI_CLUSTER_NAME}
+          - name: PATRONI_REPLICATION_USERNAME
+            value: ${PATRONI_REPLICATION_USERNAME}
+          - name: PATRONI_REPLICATION_PASSWORD
+            valueFrom:
+              secretKeyRef:
+                key: replication-password
+                name: ${PATRONI_CLUSTER_NAME}
+          - name: PATRONI_SCOPE
+            value: ${PATRONI_CLUSTER_NAME}
+          - name: PATRONI_NAME
+            valueFrom:
+              fieldRef:
+                apiVersion: v1
+                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
+          image: docker-registry.default.svc:5000/${NAMESPACE}/patroni:latest
+          imagePullPolicy: IfNotPresent
+          name: ${APPLICATION_NAME}
+          ports:
+          - containerPort: 8008
+            protocol: TCP
+          - containerPort: 5432
+            protocol: TCP
+          resources: {}
+          terminationMessagePath: /dev/termination-log
+          terminationMessagePolicy: File
+          volumeMounts:
+          - mountPath: /home/postgres/pgdata
+            name: pgdata
+        dnsPolicy: ClusterFirst
+        restartPolicy: Always
+        schedulerName: default-scheduler
+        securityContext: {}
+        serviceAccount: ${SERVICE_ACCOUNT}
+        serviceAccountName: ${SERVICE_ACCOUNT}
+        terminationGracePeriodSeconds: 0
+        volumes:
+        - name: pgdata
+          emptyDir: {}
+    updateStrategy:
+      type: OnDelete
+- apiVersion: v1
+  kind: Endpoints
+  metadata:
+    name: ${APPLICATION_NAME}
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+  subsets: []
+- apiVersion: v1
+  kind: ServiceAccount
+  metadata:
+    name: ${SERVICE_ACCOUNT}
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: Role
+  metadata:
+    name: ${SERVICE_ACCOUNT}
+  rules:
+  - apiGroups:
+    - ""
+    resources:
+    - configmaps
+    verbs:
+    - create
+    - get
+    - list
+    - patch
+    - update
+    - watch
+    # delete is required only for 'patronictl remove'
+    - delete
+  - apiGroups:
+    - ""
+    resources:
+    - endpoints
+    verbs:
+    - get
+    - patch
+    - update
+    # the following three privileges are necessary only when using endpoints
+    - create
+    - list
+    - watch
+    # delete is required only for for 'patronictl remove'
+    - delete
+  - apiGroups:
+    - ""
+    resources:
+    - pods
+    verbs:
+    - get
+    - list
+    - patch
+    - update
+    - watch
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: RoleBinding
+  metadata:
+    name: ${SERVICE_ACCOUNT}
+  roleRef:
+    apiGroup: rbac.authorization.k8s.io
+    kind: Role
+    name: ${SERVICE_ACCOUNT}
+  subjects:
+  - kind: ServiceAccount
+    name: ${SERVICE_ACCOUNT}
+parameters:
+- description: The name of the application for labelling all artifacts.
+  displayName: Application Name
+  name: APPLICATION_NAME
+  value: patroni-ephemeral
+- description: The name of the patroni-pgsql cluster.
+  displayName: Cluster Name
+  name: PATRONI_CLUSTER_NAME
+  value: patroni-ephemeral
+- description: The name of the OpenShift Service exposed for the patroni-ephemeral-master container.
+  displayName: Master service name.  
+  name: PATRONI_MASTER_SERVICE_NAME
+  value: patroni-ephemeral-master
+- description: The name of the OpenShift Service exposed for the patroni-ephemeral-replica containers.
+  displayName: Replica service name. 
+  name: PATRONI_REPLICA_SERVICE_NAME
+  value: patroni-ephemeral-replica
+- description: Maximum amount of memory the container can use.
+  displayName: Memory Limit
+  name: MEMORY_LIMIT
+  value: 512Mi
+- description: The OpenShift Namespace where the patroni and postgresql ImageStream resides.
+  displayName: ImageStream Namespace
+  name: NAMESPACE
+  value: openshift
+- description: Username of the superuser account for initialization.
+  displayName: Superuser Username
+  name: PATRONI_SUPERUSER_USERNAME
+  value: postgres
+- description: Password of the superuser account for initialization.
+  displayName: Superuser Passsword
+  name: PATRONI_SUPERUSER_PASSWORD
+  value: postgres
+- description: Username of the replication account for initialization.
+  displayName: Replication Username
+  name: PATRONI_REPLICATION_USERNAME
+  value: postgres
+- description: Password of the replication account for initialization.
+  displayName: Repication Passsword
+  name: PATRONI_REPLICATION_PASSWORD
+  value: postgres
+- description: Service account name used for pods and rolebindings to form a cluster in the project. 
+  displayName: Service Account
+  name: SERVICE_ACCOUNT
+  value: patroniocp

kubernetes/openshift-example/templates/template_patroni_persistent.yaml

@@ -0,0 +1,303 @@
+apiVersion: v1
+kind: Template
+metadata:
+  name: patroni-pgsql-persistent
+  annotations:
+    description: |-
+      Patroni Postgresql database cluster, with persistent storage.
+
+      WARNING: Any data stored will be lost upon pod destruction. Only use this template for testing.
+    iconClass: icon-postgresql
+    openshift.io/display-name: Patroni Postgresql (Persistent)
+    openshift.io/long-description: This template deploys a a patroni postgresql HA cluster without persistent storage.
+    tags: postgresql
+objects:
+- apiVersion: v1
+  kind: Service
+  metadata:
+    creationTimestamp: null
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${PATRONI_CLUSTER_NAME}
+  spec:
+    ports:
+    - port: 5432
+      protocol: TCP
+      targetPort: 5432
+    sessionAffinity: None
+    type: ClusterIP
+  status:
+    loadBalancer: {}
+- apiVersion: v1
+  kind: Service
+  metadata:
+    creationTimestamp: null
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${PATRONI_MASTER_SERVICE_NAME}
+  spec:
+    ports:
+    - port: 5432
+      protocol: TCP
+      targetPort: 5432
+    selector:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+      role: master
+    sessionAffinity: None
+    type: ClusterIP
+  status:
+    loadBalancer: {}
+- apiVersion: v1
+  kind: Secret
+  metadata:
+    name: ${PATRONI_CLUSTER_NAME}
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+  stringData:
+    superuser-password: ${PATRONI_SUPERUSER_PASSWORD}
+    replication-password: ${PATRONI_REPLICATION_PASSWORD}
+- apiVersion: v1
+  kind: Service
+  metadata:
+    creationTimestamp: null
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${PATRONI_REPLICA_SERVICE_NAME}
+  spec:
+    ports:
+    - port: 5432
+      protocol: TCP
+      targetPort: 5432
+    selector:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+      role: replica
+    sessionAffinity: None
+    type: ClusterIP
+  status:
+    loadBalancer: {}
+- apiVersion: apps/v1
+  kind: StatefulSet
+  metadata:
+    creationTimestamp: null
+    generation: 3
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+    name: ${APPLICATION_NAME}
+  spec:
+    podManagementPolicy: OrderedReady
+    replicas: 3
+    revisionHistoryLimit: 10
+    selector:
+      matchLabels:
+        application: ${APPLICATION_NAME}
+        cluster-name: ${PATRONI_CLUSTER_NAME}
+    serviceName: ${APPLICATION_NAME}
+    template:
+      metadata:
+        creationTimestamp: null
+        labels:
+          application: ${APPLICATION_NAME}
+          cluster-name: ${PATRONI_CLUSTER_NAME}
+      spec:
+        containers:
+        - env:
+          - name: PATRONI_KUBERNETES_POD_IP
+            valueFrom:
+              fieldRef:
+                apiVersion: v1
+                fieldPath: status.podIP
+          - name: PATRONI_KUBERNETES_NAMESPACE
+            valueFrom:
+              fieldRef:
+                apiVersion: v1
+                fieldPath: metadata.namespace
+          - name: PATRONI_KUBERNETES_LABELS
+            value: '{application: ${APPLICATION_NAME}, cluster-name: ${PATRONI_CLUSTER_NAME}}'
+          - name: PATRONI_SUPERUSER_USERNAME
+            value: ${PATRONI_SUPERUSER_USERNAME}
+          - name: PATRONI_SUPERUSER_PASSWORD
+            valueFrom:
+              secretKeyRef:
+                key: superuser-password
+                name: ${PATRONI_CLUSTER_NAME}
+          - name: PATRONI_REPLICATION_USERNAME
+            value: ${PATRONI_REPLICATION_USERNAME}
+          - name: PATRONI_REPLICATION_PASSWORD
+            valueFrom:
+              secretKeyRef:
+                key: replication-password
+                name: ${PATRONI_CLUSTER_NAME}
+          - name: PATRONI_SCOPE
+            value: ${PATRONI_CLUSTER_NAME}
+          - name: PATRONI_NAME
+            valueFrom:
+              fieldRef:
+                apiVersion: v1
+                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
+          image: docker-registry.default.svc:5000/${NAMESPACE}/patroni:latest
+          imagePullPolicy: IfNotPresent
+          name: ${APPLICATION_NAME}
+          ports:
+          - containerPort: 8008
+            protocol: TCP
+          - containerPort: 5432
+            protocol: TCP
+          resources: {}
+          terminationMessagePath: /dev/termination-log
+          terminationMessagePolicy: File
+          volumeMounts:
+          - mountPath: /home/postgres/pgdata
+            name: ${APPLICATION_NAME}
+        dnsPolicy: ClusterFirst
+        restartPolicy: Always
+        schedulerName: default-scheduler
+        securityContext: {}
+        serviceAccount: ${SERVICE_ACCOUNT}
+        serviceAccountName: ${SERVICE_ACCOUNT}
+        terminationGracePeriodSeconds: 0
+        volumes:
+        - name: ${APPLICATION_NAME}
+          persistentVolumeClaim: 
+            claimName: ${APPLICATION_NAME}
+    volumeClaimTemplates:
+    - metadata:
+        labels:
+          application: ${APPLICATION_NAME}
+        name: ${APPLICATION_NAME}
+      spec:
+        accessModes:
+        - ReadWriteOnce
+        resources:
+          requests:
+            storage: ${PVC_SIZE}
+    updateStrategy:
+      type: OnDelete
+- apiVersion: v1
+  kind: Endpoints
+  metadata:
+    name: ${APPLICATION_NAME}
+    labels:
+      application: ${APPLICATION_NAME}
+      cluster-name: ${PATRONI_CLUSTER_NAME}
+  subsets: []
+- apiVersion: v1
+  kind: ServiceAccount
+  metadata:
+    name: ${SERVICE_ACCOUNT}
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: Role
+  metadata:
+    name: ${SERVICE_ACCOUNT}
+  rules:
+  - apiGroups:
+    - ""
+    resources:
+    - configmaps
+    verbs:
+    - create
+    - get
+    - list
+    - patch
+    - update
+    - watch
+    # delete is required only for 'patronictl remove'
+    - delete
+  - apiGroups:
+    - ""
+    resources:
+    - endpoints
+    verbs:
+    - get
+    - patch
+    - update
+    # the following three privileges are necessary only when using endpoints
+    - create
+    - list
+    - watch
+    # delete is required only for for 'patronictl remove'
+    - delete
+  - apiGroups:
+    - ""
+    resources:
+    - pods
+    verbs:
+    - get
+    - list
+    - patch
+    - update
+    - watch
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: RoleBinding
+  metadata:
+    name: ${SERVICE_ACCOUNT}
+  roleRef:
+    apiGroup: rbac.authorization.k8s.io
+    kind: Role
+    name: ${SERVICE_ACCOUNT}
+  subjects:
+  - kind: ServiceAccount
+    name: ${SERVICE_ACCOUNT}
+parameters:
+- description: The name of the application for labelling all artifacts.
+  displayName: Application Name
+  name: APPLICATION_NAME
+  value: patroni-persistent
+- description: The name of the patroni-pgsql cluster.
+  displayName: Cluster Name
+  name: PATRONI_CLUSTER_NAME
+  value: patroni-persistent
+- description: The name of the OpenShift Service exposed for the patroni-persistent-master container.
+  displayName: Master service name.  
+  name: PATRONI_MASTER_SERVICE_NAME
+  value: patroni-persistent-master
+- description: The name of the OpenShift Service exposed for the patroni-persistent-replica containers.
+  displayName: Replica service name. 
+  name: PATRONI_REPLICA_SERVICE_NAME
+  value: patroni-persistent-replica
+- description: Maximum amount of memory the container can use.
+  displayName: Memory Limit
+  name: MEMORY_LIMIT
+  value: 512Mi
+- description: The OpenShift Namespace where the patroni and postgresql ImageStream resides.
+  displayName: ImageStream Namespace
+  name: NAMESPACE
+  value: openshift
+- description: Username of the superuser account for initialization.
+  displayName: Superuser Username
+  name: PATRONI_SUPERUSER_USERNAME
+  value: postgres
+- description: Password of the superuser account for initialization.
+  displayName: Superuser Passsword
+  name: PATRONI_SUPERUSER_PASSWORD
+  value: postgres
+- description: Username of the replication account for initialization.
+  displayName: Replication Username
+  name: PATRONI_REPLICATION_USERNAME
+  value: postgres
+- description: Password of the replication account for initialization.
+  displayName: Repication Passsword
+  name: PATRONI_REPLICATION_PASSWORD
+  value: postgres
+- description: Service account name used for pods and rolebindings to form a cluster in the project. 
+  displayName: Service Account
+  name: SERVICE_ACCOUNT
+  value: patroni-persistent
+- description: The size of the persistent volume to create. 
+  displayName: Persistent Volume Size
+  name: PVC_SIZE
+  value: 5Gi

kubernetes/openshift-example/test/Jenkinsfile

@@ -0,0 +1,43 @@
+pipeline {
+  agent any
+  stages {
+    stage ('Deploy test pod'){
+      when {
+          expression {
+            openshift.withCluster() {
+              openshift.withProject() {
+                return !openshift.selector( "dc", "pgbench" ).exists()
+              }
+            }
+          }
+        }
+      steps {
+        script {
+            openshift.withCluster() {
+                openshift.withProject() {
+                  def pgbench = openshift.newApp( "https://github.com/stewartshea/docker-pgbench/", "--name=pgbench", "-e PGPASSWORD=postgres", "-e PGUSER=postgres", "-e PGHOST=patroni-persistent-master", "-e PGDATABASE=postgres", "-e TEST_CLIENT_COUNT=20", "-e TEST_DURATION=120" )
+                  def pgbenchdc = openshift.selector( "dc", "pgbench" )
+                  timeout(5) { 
+                    pgbenchdc.rollout().status()
+                    }
+                  }
+                }
+            }
+        }
+    }
+    stage ('Run benchmark Test'){
+      steps {
+          sh '''
+          oc exec $(oc get pods -l app=pgbench | grep Running | awk '{print $1}')  ./test.sh
+          '''
+      }
+    }
+    stage ('Clean up pgtest pod'){
+      steps {
+          sh '''
+          oc delete all -l app=pgbench
+          '''
+      }
+    }
+  }
+}

kubernetes/openshift-example/test/README.md

@@ -0,0 +1,2 @@
+# Jenkins Test
+This pipeline test will create a separate deployment config for a pgbench pod and execute a test against the patroni cluster. This is a sample and should be customized. 

kubernetes/patroni_k8s.yaml

@@ -28,7 +28,7 @@ spec:
         - mountPath: /home/postgres/pgdata
           name: pgdata
         env:
-        - name: POD_IP
+        - name: PATRONI_KUBERNETES_POD_IP
           valueFrom:
             fieldRef:
               fieldPath: status.podIP
@@ -36,6 +36,8 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
+        - name: PATRONI_KUBERNETES_USE_ENDPOINTS
+          value: 'true'
         - name: PATRONI_KUBERNETES_LABELS
           value: '{application: patroni, cluster-name: patronidemo}'
         - name: PATRONI_SUPERUSER_USERNAME

patroni/dcs/consul.py

@@ -27,10 +27,14 @@ class ConsulInternalError(ConsulException):
     """An internal Consul server error occurred"""
 
 
-class InvalidSessionTTL(ConsulInternalError):
+class InvalidSessionTTL(ConsulException):
     """Session TTL is too small or too big"""
 
 
+class InvalidSession(ConsulException):
+    """invalid session"""
+
+
 class HTTPClient(object):
 
     def __init__(self, host='127.0.0.1', port=8500, token=None, scheme='http', verify=True, cert=None, ca_cert=None):
@@ -72,6 +76,8 @@ class HTTPClient(object):
             msg = '{0} {1}'.format(response.status, data)
             if data.startswith('Invalid Session TTL'):
                 raise InvalidSessionTTL(msg)
+            elif data.startswith('invalid session'):
+                raise InvalidSession(msg)
             else:
                 raise ConsulInternalError(msg)
         return base.Response(response.status, response.headers, data)
@@ -338,17 +344,15 @@ class Consul(AbstractDCS):
             logger.exception('get_cluster')
             raise ConsulError('Consul is not responding properly')
 
+    @catch_consul_errors
     def touch_member(self, data, ttl=None, permanent=False):
         cluster = self.cluster
         member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
         create_member = not permanent and self.refresh_session()
 
         if member and (create_member or member.session != self._session):
-            try:
-                self._client.kv.delete(self.member_path)
-                create_member = True
-            except Exception:
-                return False
+            self._client.kv.delete(self.member_path)
+            create_member = True
 
         if not create_member and member and deep_compare(data, member.data):
             return True
@@ -359,6 +363,9 @@ class Consul(AbstractDCS):
             if self._register_service:
                 self.update_service(not create_member and member and member.data or {}, data)
             return True
+        except InvalidSession:
+            self._session = None
+            logger.error('Our session disappeared from Consul, can not "touch_member"')
         except Exception:
             logger.exception('touch_member')
         return False
@@ -416,14 +423,21 @@ class Consul(AbstractDCS):
             return self._update_service(new_data)
 
     @catch_consul_errors
-    def _do_attempt_to_acquire_leader(self, kwargs):
-        return self.retry(self._client.kv.put, self.leader_path, self._name, **kwargs)
+    def _do_attempt_to_acquire_leader(self, permanent):
+        try:
+            kwargs = {} if permanent else {'acquire': self._session}
+            return self.retry(self._client.kv.put, self.leader_path, self._name, **kwargs)
+        except InvalidSession:
+            self._session = None
+            logger.error('Our session disappeared from Consul. Will try to get a new one and retry attempt')
+            self.refresh_session()
+            return self.retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session)
 
     def attempt_to_acquire_leader(self, permanent=False):
         if not self._session and not permanent:
             self.refresh_session()
 
-        ret = self._do_attempt_to_acquire_leader({} if permanent else {'acquire': self._session})
+        ret = self._do_attempt_to_acquire_leader(permanent)
         if not ret:
             logger.info('Could not take out TTL lock')
 
@@ -501,4 +515,5 @@ class Consul(AbstractDCS):
         try:
             return super(Consul, self).watch(None, timeout)
         finally:
+            self._last_session_refresh = 0
             self.event.clear()

patroni/dcs/zookeeper.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import select
 import time
 
 from kazoo.client import KazooClient, KazooState, KazooRetry
@@ -37,12 +38,21 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler):
                      `connect_timeout` (negotiated session timeout) as the second element."""
 
         args = list(args)
-        if len(args) == 1:
+        if len(args) == 0:  # kazoo 2.6.0 slightly changed the way how it calls create_connection method
+            kwargs['timeout'] = max(self._connect_timeout, kwargs.get('timeout', self._connect_timeout*10)/10.0)
+        elif len(args) == 1:
             args.append(self._connect_timeout)
         else:
             args[1] = max(self._connect_timeout, args[1]/10.0)
         return super(PatroniSequentialThreadingHandler, self).create_connection(*args, **kwargs)
 
+    def select(self, *args, **kwargs):
+        """Python3 raises `ValueError` if socket is closed, because fd == -1"""
+        try:
+            return super(PatroniSequentialThreadingHandler, self).select(*args, **kwargs)
+        except ValueError as e:
+            raise select.error(9, str(e))
+
 
 class ZooKeeper(AbstractDCS):
 

patroni/postgresql.py

@@ -1243,7 +1243,8 @@ class Postgresql(object):
     @contextmanager
     def _get_replication_connection_cursor(self, host='localhost', port=5432, database=None, **kwargs):
         database = database or self._database
-        with self._get_connection_cursor(host=host, port=int(port), database=database, replication='database',
+        replication = 'database' if self._major_version >= 90400 else 1
+        with self._get_connection_cursor(host=host, port=int(port), database=database, replication=replication,
                                          user=self._replication['username'], password=self._replication['password'],
                                          connect_timeout=3, options='-c statement_timeout=2000') as cur:
             yield cur

patroni/version.py

@@ -1 +1 @@
-__version__ = '1.5.1'
+__version__ = '1.5.3'

tests/test_consul.py

@@ -4,7 +4,7 @@ import unittest
 from consul import ConsulException, NotFound
 from mock import Mock, patch
 from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \
-                                ConsulError, HTTPClient, InvalidSessionTTL
+                                ConsulError, HTTPClient, InvalidSessionTTL, InvalidSession
 from test_etcd import SleepException
 
 
@@ -52,6 +52,8 @@ class TestHTTPClient(unittest.TestCase):
         self.assertRaises(ConsulInternalError, self.client.get, Mock(), '')
         self.client.http.request.return_value.data = b"Invalid Session TTL '3000000000', must be between [10s=24h0m0s]"
         self.assertRaises(InvalidSessionTTL, self.client.get, Mock(), '')
+        self.client.http.request.return_value.data = b"invalid session '16492f43-c2d6-5307-432f-e32d6f7bcbd0'"
+        self.assertRaises(InvalidSession, self.client.get, Mock(), '')
 
     def test_unknown_method(self):
         try:
@@ -110,19 +112,18 @@ class TestConsul(unittest.TestCase):
         self.c._session = 'fd4f44fe-2cac-bba5-a60b-304b51ff39b8'
         self.assertIsInstance(self.c.get_cluster(), Cluster)
 
-    @patch.object(consul.Consul.KV, 'delete', Mock(side_effect=[ConsulException, True, True]))
-    @patch.object(consul.Consul.KV, 'put', Mock(side_effect=[True, ConsulException]))
+    @patch.object(consul.Consul.KV, 'delete', Mock(side_effect=[ConsulException, True, True, True]))
+    @patch.object(consul.Consul.KV, 'put', Mock(side_effect=[True, ConsulException, InvalidSession]))
     def test_touch_member(self):
-        self.c._register_service = True
-        self.c.refresh_session = Mock(return_value=True)
-        self.c.touch_member({'balbla': 'blabla'})
-        self.c.touch_member({'balbla': 'blabla'})
-        self.c.touch_member({'balbla': 'blabla'})
         self.c.refresh_session = Mock(return_value=False)
         self.c.touch_member({'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
                              'api_url': 'http://127.0.0.1:8009/patroni'})
+        self.c._register_service = True
+        self.c.refresh_session = Mock(return_value=True)
+        for _ in range(0, 4):
+            self.c.touch_member({'balbla': 'blabla'})
 
-    @patch.object(consul.Consul.KV, 'put', Mock(return_value=False))
+    @patch.object(consul.Consul.KV, 'put', Mock(side_effect=InvalidSession))
     def test_take_leader(self):
         self.c.set_ttl(20)
         self.c.refresh_session = Mock()

tests/test_zookeeper.py

@@ -1,3 +1,4 @@
+import select
 import six
 import unittest
 
@@ -113,6 +114,11 @@ class TestPatroniSequentialThreadingHandler(unittest.TestCase):
     def test_create_connection(self):
         self.assertIsNotNone(self.handler.create_connection(()))
         self.assertIsNotNone(self.handler.create_connection((), 40))
+        self.assertIsNotNone(self.handler.create_connection(timeout=40))
+
+    @patch.object(SequentialThreadingHandler, 'select', Mock(side_effect=ValueError))
+    def test_select(self):
+        self.assertRaises(select.error, self.handler.select)
 
 
 class TestZooKeeper(unittest.TestCase):