Bumped version to 1.2.5

When all etcd servers refuse connections during watch the call will fail with an exception and will be immediately retried. This creates a huge amount of log spam potentially creating additional issues on top of losing the DCS. This patch takes note if etcd failures are repeating and starting from the second failure will sleep for a second before retrying. It additionally omits the stack trace after the first failure in a streak of failures.

.travis.yml

@@ -5,7 +5,7 @@ python:
 - "3.4"  # 2.7 and 3.5 are preinstalled by default
 env:
   global:
-  - ETCDVERSION=3.0.15 ZKVERSION=3.4.9 CONSULVERSION=0.7.2
+  - ETCDVERSION=3.0.17 ZKVERSION=3.4.9 CONSULVERSION=0.7.4
   - PYVERSIONS="2.7 3.4 3.5"
   matrix:
   - TEST_SUITE="python setup.py"

Dockerfile

@@ -1,33 +1,19 @@
 ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine
 ## It has all the necessary components to play/debug with a single node appliance, running etcd
-FROM ubuntu:16.04
+FROM postgres:9.6
 MAINTAINER Alexander Kukushkin <[email protected]>
 
-RUN echo 'APT::Install-Recommends "0";' > /etc/apt/apt.conf.d/01norecommend \
-    && echo 'APT::Install-Suggests "0";' >> /etc/apt/apt.conf.d/01norecommend
-
-ENV PGVERSION 9.6
-ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH
-RUN apt-get update -y \
+RUN 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 curl jq haproxy zookeeper python-psycopg2 python-yaml python-requests \
-        python-six python-click python-dateutil python-tzlocal python-urllib3 python-dnspython \
+    && apt-get install -y curl jq haproxy python-psycopg2 python-yaml python-requests \
+        python-six python-dateutil python-urllib3 python-dnspython \
         python-pip python-setuptools python-kazoo python-prettytable python-wheel python \
 
-    && export DISTRIB_CODENAME=$(sed -n 's/DISTRIB_CODENAME=//p' /etc/lsb-release) \
-    && echo "deb http://apt.postgresql.org/pub/repos/apt/ ${DISTRIB_CODENAME}-pgdg main" > /etc/apt/sources.list.d/pgdg.list \
-    && echo "deb-src http://apt.postgresql.org/pub/repos/apt/ ${DISTRIB_CODENAME}-pgdg main" >> /etc/apt/sources.list.d/pgdg.list \
-    && curl -s -o - https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add - \
+    && pip install python-etcd==0.4.3 python-consul==0.7.0 click tzlocal --upgrade \
 
-    ## 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 \
-
-    && apt-get update -y \
-    && apt-get install -y postgresql-contrib-${PGVERSION} \
-
-    # Remove the default cluster, which Debian stupidly starts right after installation of the packages
-    && pg_dropcluster --stop ${PGVERSION} main \
-    && pip install python-etcd==0.4.3 python-consul==0.6.1 --upgrade \
+    && mkdir -p /home/postgres \
+    && chown postgres:postgres /home/postgres \
 
     # Clean up
     && apt-get remove -y python-pip python-setuptools \
@@ -35,7 +21,7 @@ RUN apt-get update -y \
     && apt-get clean -y \
     && rm -rf /var/lib/apt/lists/* /root/.cache
 
-ENV ETCDVERSION 3.0.15
+ENV ETCDVERSION 3.1.2
 RUN curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz \
     | tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl
 
@@ -50,8 +36,7 @@ RUN ln -s /patronictl.py /usr/local/bin/patronictl
 
 ### Setting up a simple script that will serve as an entrypoint
 RUN mkdir /data/ && touch /pgpass /patroni.yml \
-    && chown postgres:postgres -R /patroni/ /data/ /pgpass /patroni.yml /etc/haproxy /var/run/ /var/lib/ /var/log/ \
-    && echo 1 > /etc/zookeeper/conf/myid
+    && chown postgres:postgres -R /patroni/ /data/ /pgpass /patroni.yml /etc/haproxy /var/run/ /var/lib/ /var/log/
 
 EXPOSE 2379 5432 8008
 

README.rst

@@ -3,7 +3,7 @@
 Patroni: A Template for PostgreSQL HA with ZooKeeper, etcd or Consul
 ------------------------------------------------------------
 
-You can find a version of this documentation that is searchable and also easier to navigagate at `patroni.readthedocs.io <https://patroni.readthedocs.io>`__.
+You can find a version of this documentation that is searchable and also easier to navigate at `patroni.readthedocs.io <https://patroni.readthedocs.io>`__.
 
 
 There are many ways to run high availability with PostgreSQL; for a list, see the `PostgreSQL Documentation <https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling>`__.
@@ -12,7 +12,7 @@ Patroni is a template for you to create your own customized, high-availability s
 
 We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely.
 
-**Note to Kubernetes users**: We're currently developing Patroni to be as useful as possible for teams running Kubernetes on top of Google Compute Engine; Patroni can be the HA solution for Postgres in such an environment. To this end, we've created a `Helm Chart <https://github.com/kubernetes/charts/tree/master/incubator/patroni>`__ that enables you to deploy a five-node Patroni cluster using a Kubernetes PetSet.
+**Note to Kubernetes users**: We're currently developing Patroni to be as useful as possible for teams running Kubernetes on top of Google Compute Engine; Patroni can be the HA solution for Postgres in such an environment. To this end, there is a `Helm chart <https://github.com/kubernetes/charts/tree/master/incubator/patroni>`__ that uses Patroni and  `Spilo  <https://github.com/zalando/spilo/>`__ to provision a five-node PostgreSQL HA cluster in a Kubernetes+GCE environment. (The Helm chart deploys Spilo Docker images, not just "bare" Patroni.)
 
 .. contents::
     :local:
@@ -29,6 +29,7 @@ For an example of a Docker-based deployment with Patroni, see `Spilo <https://gi
 
 For additional background info, see:
 
+* `Elephants on Automatic: HA Clustered PostgreSQL with Helm <https://www.youtube.com/watch?v=CftcVhFMGSY>`_, talk by Josh Berkus and Oleksii Kliukin at KubeCon Berlin 2017 
 * `PostgreSQL HA with Kubernetes and Patroni <https://www.youtube.com/watch?v=iruaCgeG7qs>`__, talk by Josh Berkus at KubeCon 2016 (video)
 * `Feb. 2016 Zalando Tech blog post <https://tech.zalando.de/blog/zalandos-patroni-a-template-for-high-availability-postgresql/>`__
 
@@ -95,7 +96,7 @@ Go `here <https://github.com/zalando/patroni/blob/master/docs/ENVIRONMENT.rst>`_
 Replication Choices
 ===============
 
-Patroni uses Postgres' streaming replication, which is asynchronous by default. Patroni's asynchronous replication configuration allows for ``maximum_lag_on_failover`` settings. This setting ensures failover will not occur if a follower is more than a certain number of bytes behind the leader. This setting should be increased or decreased based on business requirements. It's also possible to use synchronous replication for better durability guarantees. See `replication modes documentation <https://github.com/zalando/patroni/blob/master/docs/replication_modes.rst>` for details.
+Patroni uses Postgres' streaming replication, which is asynchronous by default. Patroni's asynchronous replication configuration allows for ``maximum_lag_on_failover`` settings. This setting ensures failover will not occur if a follower is more than a certain number of bytes behind the leader. This setting should be increased or decreased based on business requirements. It's also possible to use synchronous replication for better durability guarantees. See `replication modes documentation <https://github.com/zalando/patroni/blob/master/docs/replication_modes.rst>`__ for details.
 
 ===============================
 Applications Should Not Use Superusers

docker-compose.yml

@@ -50,8 +50,8 @@ haproxy:
     links:
         - patroni_etcd:patroni_etcd
     ports:
-        - "5000"
-        - "5001"
+        - "5000:5000"
+        - "5001:5001"
     environment:
         PATRONI_ETCD_HOST: patroni_etcd:2379
         PATRONI_SCOPE: testcluster

patroni/dcs/etcd.py

@@ -296,19 +296,6 @@ class Client(etcd.Client):
         self._machines_cache_updated = time.time()
 
 
-def catch_etcd_errors(func):
-    def wrapper(*args, **kwargs):
-        try:
-            return func(*args, **kwargs) is not None
-        except (RetryFailedError, etcd.EtcdException):
-            return False
-        except:
-            logger.exception("")
-            raise EtcdError("unexpected error")
-
-    return wrapper
-
-
 class Etcd(AbstractDCS):
 
     def __init__(self, config):
@@ -320,10 +307,36 @@ class Etcd(AbstractDCS):
                                               etcd.EtcdEventIndexCleared))
         self._client = self.get_etcd_client(config)
         self.__do_not_watch = False
+        self._has_failed = False
 
     def retry(self, *args, **kwargs):
         return self._retry.copy()(*args, **kwargs)
 
+    def _handle_exception(self, e, name='', do_sleep=False, raise_ex=None):
+        if not self._has_failed:
+            logger.exception(name)
+        else:
+            logger.error(e)
+            if do_sleep:
+                time.sleep(1)
+        self._has_failed = True
+        if isinstance(raise_ex, Exception):
+            raise raise_ex
+
+    def catch_etcd_errors(func):
+        def wrapper(self, *args, **kwargs):
+            try:
+                retval = func(self, *args, **kwargs) is not None
+                self._has_failed = False
+                return retval
+            except (RetryFailedError, etcd.EtcdException) as e:
+                self._handle_exception(e)
+                return False
+            except Exception as e:
+                self._handle_exception(e, raise_ex=EtcdError('unexpected error'))
+
+        return wrapper
+
     @staticmethod
     def get_etcd_client(config):
         if 'proxy' in config:
@@ -448,9 +461,9 @@ class Etcd(AbstractDCS):
             self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync)
         except etcd.EtcdKeyNotFound:
             self._cluster = Cluster(None, None, None, None, [], None, None)
-        except:
-            logger.exception('get_cluster')
-            raise EtcdError('Etcd is not responding properly')
+        except Exception as e:
+            self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
+        self._has_failed = False
 
     @catch_etcd_errors
     def touch_member(self, data, ttl=None, permanent=False):
@@ -524,14 +537,19 @@ class Etcd(AbstractDCS):
             while timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
                 try:
                     self._client.watch(self.leader_path, index=leader_index, timeout=timeout + 0.5)
+                    self._has_failed = False
                     # Synchronous work of all cluster members with etcd is less expensive
                     # than reestablishing http connection every time from every replica.
                     return True
                 except etcd.EtcdWatchTimedOut:
                     self._client.http.clear()
+                    self._has_failed = False
                     return False
-                except etcd.EtcdException:
-                    logger.exception('watch')
+                except (etcd.EtcdEventIndexCleared, etcd.EtcdWatcherCleared):  # Watch failed
+                    self._has_failed = False
+                    return True  # leave the loop, because watch with the same parameters will fail anyway
+                except etcd.EtcdException as e:
+                    self._handle_exception(e, 'watch', True)
 
                 timeout = end_time - time.time()
 

patroni/ha.py

@@ -886,6 +886,11 @@ class Ha(object):
 
             # is data directory empty?
             if self.state_handler.data_directory_empty():
+                # is this instance the leader?
+                if self.has_lock():
+                    self.release_leader_key_voluntarily()
+                    return 'released leader key voluntarily as data dir empty and currently leader'
+
                 return self.bootstrap()  # new node
             # "bootstrap", but data directory is not empty
             elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked() and not self.is_paused():

patroni/postgresql.py

@@ -116,6 +116,7 @@ class Postgresql(object):
         self._trigger_file = config.get('recovery_conf', {}).get('trigger_file') or 'promote'
         self._trigger_file = os.path.abspath(os.path.join(self._data_dir, self._trigger_file))
 
+        self._connection_lock = Lock()
         self._connection = None
         self._cursor_holder = None
         self._sysid = None
@@ -177,6 +178,8 @@ class Postgresql(object):
                 parameters.pop('synchronous_standby_names', None)
             else:
                 parameters['synchronous_standby_names'] = self._synchronous_standby_names
+        if self._major_version >= 9.6 and parameters['wal_level'] == 'hot_standby':
+            parameters['wal_level'] = 'replica'
         return {k: v for k, v in parameters.items() if not self._major_version or
                 self._major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 9.1))[2]}
 
@@ -249,8 +252,6 @@ class Postgresql(object):
                     elif r[0] in changes:
                         unit = changes['wal_segment_size'] if r[0] in ('min_wal_size', 'max_wal_size') else r[2]
                         new_value = changes.pop(r[0])
-                        if self._major_version >= 9.6 and r[0] == 'wal_level' and new_value == 'hot_standby':
-                            new_value = 'replica'
                         if new_value is None or not compare_values(r[3], unit, r[1], new_value):
                             if r[4] == 'postmaster':
                                 pending_restart = True
@@ -351,10 +352,11 @@ class Postgresql(object):
         return ret
 
     def connection(self):
-        if not self._connection or self._connection.closed != 0:
-            self._connection = psycopg2.connect(**self._local_connect_kwargs)
-            self._connection.autocommit = True
-            self.server_version = self._connection.server_version
+        with self._connection_lock:
+            if not self._connection or self._connection.closed != 0:
+                self._connection = psycopg2.connect(**self._local_connect_kwargs)
+                self._connection.autocommit = True
+                self.server_version = self._connection.server_version
         return self._connection
 
     def _cursor(self):
@@ -435,6 +437,7 @@ class Postgresql(object):
         if ret:
             self.write_pg_hba(config.get('pg_hba', []))
             self._major_version = self.get_major_version()
+            self._server_parameters = self.get_server_parameters(self.config)
         else:
             self.set_state('initdb failed')
         return ret
@@ -697,9 +700,7 @@ class Postgresql(object):
         self._write_postgresql_conf()
         self.resolve_connection_addresses()
 
-        opts = {p: self._server_parameters[p] for p, v in self.CMDLINE_OPTIONS.items() if self._major_version >= v[2]}
-        if self._major_version >= 9.6 and opts['wal_level'] == 'hot_standby':
-            opts['wal_level'] = 'replica'
+        opts = {p: self._server_parameters[p] for p in self.CMDLINE_OPTIONS if p in self._server_parameters}
         options = ['--{0}={1}'.format(p, v) for p, v in opts.items()]
 
         start_initiated = time.time()
@@ -1180,9 +1181,12 @@ $$""".format(name, ' '.join(options)), name, password, password)
 
                 # drop unused slots
                 for slot in set(self._replication_slots) - slots:
-                    self._query("""SELECT pg_drop_replication_slot(%s)
-                                    WHERE EXISTS(SELECT 1 FROM pg_replication_slots
-                                    WHERE slot_name = %s AND NOT active)""", slot, slot)
+                    cursor = self._query("""SELECT pg_drop_replication_slot(%s)
+                                             WHERE EXISTS(SELECT 1 FROM pg_replication_slots
+                                             WHERE slot_name = %s AND NOT active)""", slot, slot)
+
+                    if cursor.rowcount != 1:  # Either slot doesn't exists or it is still active
+                        self._schedule_load_slots = True  # schedule load_replication_slots on the next iteration
 
                 # create new slots
                 for slot in slots - set(self._replication_slots):
@@ -1209,6 +1213,7 @@ $$""".format(name, ' '.join(options)), name, password, password)
         ret = self.create_replica(clone_member) == 0
         if ret:
             self._major_version = self.get_major_version()
+            self._server_parameters = self.get_server_parameters(self.config)
             self.delete_trigger_file()
             self.restore_configuration_files()
         return ret

patroni/scripts/aws.py

@@ -6,13 +6,18 @@ from requests.exceptions import RequestException
 import sys
 import boto.ec2
 
+from patroni.utils import Retry, RetryFailedError
+
 logger = logging.getLogger(__name__)
 
+retry_timeout = 15
+
 
 class AWSConnection(object):
     def __init__(self, cluster_name):
         self.available = False
         self.cluster_name = cluster_name if cluster_name is not None else 'unknown'
+        self._retry = Retry(deadline=retry_timeout, max_delay=5, max_tries=-1, retry_exceptions=(boto.exception,))
         try:
             # get the instance id
             r = requests.get('http://169.254.169.254/latest/dynamic/instance-identity/document', timeout=0.1)
@@ -29,40 +34,36 @@ class AWSConnection(object):
                 return
             self.available = True
 
+    def retry(self, *args, **kwargs):
+        return self._retry.copy()(*args, **kwargs)
+
     def aws_available(self):
         return self.available
 
-    def _tag_ebs(self, role):
+    def _tag_ebs(self, conn, role):
         """ set tags, carrying the cluster name, instance role and instance id for the EBS storage """
-        if not self.available:
-            return False
-
         tags = {'Name': 'spilo_' + self.cluster_name, 'Role': role, 'Instance': self.instance_id}
-        try:
-            conn = boto.ec2.connect_to_region(self.region)
-            volumes = conn.get_all_volumes(filters={'attachment.instance-id': self.instance_id})
-            conn.create_tags([v.id for v in volumes], tags)
-        except Exception as e:
-            logger.info('could not set tags for EBS storage devices attached: {}'.format(e))
-            return False
-        return True
+        volumes = conn.get_all_volumes(filters={'attachment.instance-id': self.instance_id})
+        conn.create_tags([v.id for v in volumes], tags)
 
-    def _tag_ec2(self, role):
+    def _tag_ec2(self, conn, role):
         """ tag the current EC2 instance with a cluster role """
-        if not self.available:
-            return False
         tags = {'Role': role}
-        try:
-            conn = boto.ec2.connect_to_region(self.region)
-            conn.create_tags([self.instance_id], tags)
-        except Exception as e:
-            logger.info("could not set tags for EC2 instance %s: %s", self.instance_id, e)
-            return False
-        return True
+        conn.create_tags([self.instance_id], tags)
 
     def on_role_change(self, new_role):
-        ret = self._tag_ec2(new_role)
-        return self._tag_ebs(new_role) and ret
+        if not self.available:
+            return False
+        try:
+            conn = self.retry(boto.ec2.connect_to_region, self.region)
+            self.retry(self._tag_ec2, conn, new_role)
+            self.retry(self._tag_ebs, conn, new_role)
+        except RetryFailedError:
+            logger.warning("Unable to communicate to AWS "
+                           "when setting tags for the EC2 instance {0} "
+                           "and attached EBS volumes".format(self.instance_id))
+            return False
+        return True
 
 
 def main():

patroni/scripts/wale_restore.py

@@ -133,8 +133,16 @@ class WALERestore(object):
                     with psycopg2.connect(self.master_connection) as con:
                         con.autocommit = True
                         with con.cursor() as cur:
-                            cur.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)",
-                                        (backup_start_lsn,))
+                            cur.execute("""SELECT CASE WHEN pg_is_in_recovery()
+                                                       THEN GREATEST(
+                                                                pg_xlog_location_diff(COALESCE(
+                                                                  pg_last_xlog_receive_location(), '0/0'), %s)::bigint,
+                                                                pg_xlog_location_diff(
+                                                                  pg_last_xlog_replay_location(), %s)::bigint)
+                                                       ELSE pg_xlog_location_diff(
+                                                                pg_current_xlog_location(), %s)::bigint
+                                                   END""", (backup_start_lsn, backup_start_lsn, backup_start_lsn))
+
                             diff_in_bytes = int(cur.fetchone()[0])
                 except psycopg2.Error:
                     logger.exception('could not determine difference with the master location')

patroni/version.py

@@ -1 +1 @@
-__version__ = '1.2.3'
+__version__ = '1.2.5'

tests/test_api.py

@@ -101,6 +101,8 @@ class MockRequest(object):
     def makefile(self, *args, **kwargs):
         return IO(self.request)
 
+    def sendall(self, *args, **kwargs):
+        pass
 
 class MockRestApiServer(RestApiServer):
 

tests/test_aws.py

@@ -6,6 +6,7 @@ import unittest
 from mock import Mock, patch
 from collections import namedtuple
 from patroni.scripts.aws import AWSConnection, main as _main
+from patroni.utils import RetryFailedError
 from requests.exceptions import RequestException
 
 
@@ -16,13 +17,13 @@ class MockEc2Connection(object):
 
     def get_all_volumes(self, filters):
         if self.error:
-            raise Exception("get_all_volumes")
+            raise boto.exception("get_all_volumes")
         oid = namedtuple('Volume', 'id')
         return [oid(id='a'), oid(id='b')]
 
     def create_tags(self, objects, tags):
         if self.error or len(objects) == 0:
-            raise Exception("create_tags")
+            raise boto.exception("create_tags")
         return True
 
 
@@ -63,30 +64,20 @@ class TestAWSConnection(unittest.TestCase):
         self.assertTrue(self.conn.aws_available())
 
     def test_on_role_change(self):
-        self.assertTrue(self.conn._tag_ebs('master'))
-        self.assertTrue(self.conn._tag_ec2('master'))
         self.assertTrue(self.conn.on_role_change('master'))
+        self.conn.retry = Mock(side_effect=RetryFailedError("retry failed"))
+        self.assertFalse(self.conn.on_role_change('master'))
 
     def test_non_aws(self):
         self.error = True
         conn = AWSConnection('test')
-        self.assertFalse(conn.aws_available())
-        self.assertFalse(conn._tag_ebs('master'))
-        self.assertFalse(conn._tag_ec2('master'))
+        self.assertFalse(conn.on_role_change("master"))
 
     def test_aws_bizare_response(self):
         self.json_error = True
         conn = AWSConnection('test')
         self.assertFalse(conn.aws_available())
 
-    def test_aws_tag_ebs_error(self):
-        self.error = True
-        self.assertFalse(self.conn._tag_ebs("master"))
-
-    def test_aws_tag_ec2_error(self):
-        self.error = True
-        self.assertFalse(self.conn._tag_ec2("master"))
-
     @patch('sys.exit', Mock())
     def test_main(self):
         self.assertIsNone(_main())

tests/test_etcd.py

@@ -59,8 +59,10 @@ def etcd_watch(self, key, index=None, timeout=None, recursive=None):
         raise etcd.EtcdWatchTimedOut
     elif timeout == 5.0:
         return etcd.EtcdResult('delete', {})
-    elif timeout == 10.0:
+    elif 5 < timeout <= 10.0:
         raise etcd.EtcdException
+    elif timeout == 20.0:
+        raise etcd.EtcdEventIndexCleared
 
 
 def etcd_write(self, key, value, **kwargs):
@@ -300,6 +302,7 @@ class TestEtcd(unittest.TestCase):
     def test_delete_cluster(self):
         self.assertFalse(self.etcd.delete_cluster())
 
+    @patch('time.sleep', Mock(side_effect=SleepException))
     @patch.object(etcd.Client, 'watch', etcd_watch)
     def test_watch(self):
         self.etcd.watch(None, 0)
@@ -307,7 +310,8 @@ class TestEtcd(unittest.TestCase):
         self.etcd.watch(20729, 1.5)
         self.etcd.watch(20729, 4.5)
         with patch.object(AbstractDCS, 'watch', Mock()):
-            self.etcd.watch(20729, 9.5)
+            self.assertTrue(self.etcd.watch(20729, 19.5))
+            self.assertRaises(SleepException, self.etcd.watch, 20729, 9.5)
 
     def test_other_exceptions(self):
         self.etcd.retry = Mock(side_effect=AttributeError('foo'))

tests/test_ha.py

@@ -771,3 +771,14 @@ class TestHa(unittest.TestCase):
 
     def test_wakup(self):
         self.ha.wakeup()
+
+    def test_leader_with_empty_directory(self):
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.ha.has_lock = true
+        self.p.data_directory_empty = true
+        self.assertEquals(self.ha.run_cycle(), 'released leader key voluntarily as data dir empty and currently leader')
+
+        # as has_lock is mocked out, we need to fake the leader key release
+        self.ha.has_lock = false
+        # will not say bootstrap from leader as replica can't self elect
+        self.assertEquals(self.ha.run_cycle(), "trying to bootstrap from replica 'other'")

tests/test_postgresql.py

@@ -19,6 +19,7 @@ class MockCursor(object):
     def __init__(self, connection):
         self.connection = connection
         self.closed = False
+        self.rowcount = 0
         self.results = []
 
     def execute(self, sql, *params):
@@ -41,8 +42,7 @@ class MockCursor(object):
                             ('search_path', 'public', None, 'string', 'user'),
                             ('port', '5433', None, 'integer', 'postmaster'),
                             ('listen_addresses', '*', None, 'string', 'postmaster'),
-                            ('autovacuum', 'on', None, 'bool', 'sighup'),
-                            ('wal_level', 'replica', None, 'enum', 'postmaster')]
+                            ('autovacuum', 'on', None, 'bool', 'sighup')]
         else:
             self.results = [(None, None, None, None, None, None, None, None, None, None)]
 
@@ -766,7 +766,7 @@ class TestPostgresql(unittest.TestCase):
         self.assertEquals(value_in_conf(), None)
 
     def test_get_server_parameters(self):
-        config = {'synchronous_mode': True, 'parameters': {}, 'listen': '0'}
+        config = {'synchronous_mode': True, 'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'}
         self.p.get_server_parameters(config)
         self.p.set_synchronous_standby('foo')
         self.p.get_server_parameters(config)