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.

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 \
+RUN echo 'APT::Install-Recommends "0";\nAPT::Install-Suggests "0";' > /etc/apt/apt.conf.d/01norecommend \
-    && echo 'APT::Install-Suggests "0";' >> /etc/apt/apt.conf.d/01norecommend
+    && apt-get update -y \
-
-ENV PGVERSION 9.6
-ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH
-RUN apt-get update -y \
     && apt-get upgrade -y \
-    && apt-get install -y curl jq haproxy zookeeper python-psycopg2 python-yaml python-requests \
+    && apt-get install -y curl jq haproxy python-psycopg2 python-yaml python-requests \
-        python-six python-click python-dateutil python-tzlocal python-urllib3 python-dnspython \
+        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) \
+    && pip install python-etcd==0.4.3 python-consul==0.7.0 click tzlocal --upgrade \
-    && 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 - \
 
-    ## Make sure we have a en_US.UTF-8 locale available
+    && mkdir -p /home/postgres \
-    && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \
+    && chown postgres:postgres /home/postgres \
-
-    && 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 \
 
     # 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/ \
+    && chown postgres:postgres -R /patroni/ /data/ /pgpass /patroni.yml /etc/haproxy /var/run/ /var/lib/ /var/log/
-    && echo 1 > /etc/zookeeper/conf/myid
 
 EXPOSE 2379 5432 8008
 

README.rst

@@ -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/>`__
 

docker-compose.yml

@@ -50,8 +50,8 @@ haproxy:
     links:
         - patroni_etcd:patroni_etcd
     ports:
-        - "5000"
+        - "5000:5000"
-        - "5001"
+        - "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:
+        except Exception as e:
-            logger.exception('get_cluster')
+            self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
-            raise EtcdError('Etcd is not responding properly')
+        self._has_failed = False
 
     @catch_etcd_errors
     def touch_member(self, data, ttl=None, permanent=False):
@@ -524,16 +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.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:
+                except etcd.EtcdException as e:
-                    logger.exception('watch')
+                    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
@@ -351,10 +352,11 @@ class Postgresql(object):
         return ret
 
     def connection(self):
-        if not self._connection or self._connection.closed != 0:
+        with self._connection_lock:
-            self._connection = psycopg2.connect(**self._local_connect_kwargs)
+            if not self._connection or self._connection.closed != 0:
-            self._connection.autocommit = True
+                self._connection = psycopg2.connect(**self._local_connect_kwargs)
-            self.server_version = self._connection.server_version
+                self._connection.autocommit = True
+                self.server_version = self._connection.server_version
         return self._connection
 
     def _cursor(self):
@@ -1179,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)
+                    cursor = self._query("""SELECT pg_drop_replication_slot(%s)
-                                    WHERE EXISTS(SELECT 1 FROM pg_replication_slots
+                                             WHERE EXISTS(SELECT 1 FROM pg_replication_slots
-                                    WHERE slot_name = %s AND NOT active)""", slot, slot)
+                                             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):

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:
+        volumes = conn.get_all_volumes(filters={'attachment.instance-id': self.instance_id})
-            conn = boto.ec2.connect_to_region(self.region)
+        conn.create_tags([v.id for v in volumes], tags)
-            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
 
-    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.create_tags([self.instance_id], tags)
-            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
 
     def on_role_change(self, new_role):
-        ret = self._tag_ec2(new_role)
+        if not self.available:
-        return self._tag_ebs(new_role) and ret
+            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/version.py

@@ -1 +1 @@
-__version__ = '1.2.4'
+__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.on_role_change("master"))
-        self.assertFalse(conn._tag_ebs('master'))
-        self.assertFalse(conn._tag_ec2('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,7 +59,7 @@ 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
@@ -302,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)
@@ -310,7 +311,7 @@ class TestEtcd(unittest.TestCase):
         self.etcd.watch(20729, 4.5)
         with patch.object(AbstractDCS, 'watch', Mock()):
             self.assertTrue(self.etcd.watch(20729, 19.5))
-            self.etcd.watch(20729, 9.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):