From 793325cb609010cfc195986f9cf1ea7f2c9fbc09 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Wed, 23 Sep 2015 18:38:17 +0200
Subject: [PATCH 01/16] add support for pg_rewind.

---
 patroni/postgresql.py | 41 +++++++++++++++++++++++++++++++++++++----
 postgres0.yml         |  5 +++++
 postgres1.yml         |  5 +++++
 3 files changed, 47 insertions(+), 4 deletions(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 532e65c8..b69099a9 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -47,6 +47,7 @@ class Postgresql:
         self.replication = config['replication']
         self.superuser = config['superuser']
         self.admin = config['admin']
+        self.pg_rewind = config.get('pg_rewind', {})
         self.callback = config.get('callbacks', {})
         self.use_slots = config.get('use_slots', True)
         self.schedule_load_slots = self.use_slots
@@ -69,6 +70,17 @@ class Postgresql:
         self._cursor_holder = None
         self.members = []  # list of already existing replication slots
         self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException)
+        try:
+            self._pg_rewind_present = ('username' in self.pg_rewind and
+                                       ('wal_log_hints' in self.config['parameters'] or
+                                        'data_checksums' in self.config['parameters']) and
+                                       os.system("pg_rewind --version >/dev/null 2>&1") == 0)
+            if self._pg_rewind_present:
+                self.pg_rewind['user'] = self.pg_rewind['username']
+        except:
+            self._pg_rewind_present = False
+        if self.pg_rewind and not self._pg_rewind_present:
+            logger.warning("pg_rewind support is disabled")
 
     def get_local_address(self):
         listen_addresses = self.listen_addresses.split(',')
@@ -265,8 +277,10 @@ class Postgresql:
                 f.write(line + '\n')
 
     @staticmethod
-    def primary_conninfo(leader_url):
+    def primary_conninfo(leader_url, replacement=None):
         r = parseurl(leader_url)
+        if replacement is not None:
+            r.update(replacement)
         return 'user={user} password={password} host={host} port={port} sslmode=prefer sslcompression=1'.format(**r)
 
     def check_recovery_conf(self, leader):
@@ -299,9 +313,28 @@ recovery_target_timeline = 'latest'
     def follow_the_leader(self, leader):
         if not self.check_recovery_conf(leader):
             self.write_recovery_conf(leader)
-            run_callback = self.role == 'master'
-            self.restart()
-            run_callback and self.call_nowait(ACTION_ON_ROLE_CHANGE)
+            change_role = self.role == 'master'
+
+            if leader and change_role and self._pg_rewind_present:
+                self.stop()
+                pc = self.primary_conninfo(leader.conn_url,
+                                           self.pg_rewind) + ' dbname=postgres'
+                logger.info("running pg_rewind from {}".format(pc))
+                pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
+                try:
+                    ret = (subprocess.call(pg_rewind) == 0)
+                except:
+                    ret = False
+                # pg_rewind removes recovery.conf, we have to reinstate it.
+                if ret:
+                    self.write_recovery_conf(leader)
+                    self.start()
+                else:
+                    self.remove_data_directory()
+                    logger.error("unable to rewind the former leader")
+            else:
+                ret = self.restart()
+            change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE)
 
     def save_configuration_files(self):
         """
diff --git a/postgres0.yml b/postgres0.yml
index f800183a..a155b1cd 100644
--- a/postgres0.yml
+++ b/postgres0.yml
@@ -34,6 +34,9 @@ postgresql:
   data_dir: data/postgresql0
   maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
   use_slots: True
+  pg_rewind:
+    username: postgres
+    password: zalando
   pg_hba:
   - host all all 0.0.0.0/0 md5
   - hostssl all all 0.0.0.0/0 md5
@@ -42,6 +45,7 @@ postgresql:
     password: rep-pass
     network:  127.0.0.1/32
   superuser:
+    username: postgres
     password: zalando
   admin:
     username: admin
@@ -62,3 +66,4 @@ postgresql:
     archive_timeout: 1800s
     max_replication_slots: 5
     hot_standby: "on"
+    wal_log_hints: "on"
diff --git a/postgres1.yml b/postgres1.yml
index e1c3e663..94e33a42 100644
--- a/postgres1.yml
+++ b/postgres1.yml
@@ -34,6 +34,9 @@ postgresql:
   data_dir: data/postgresql1
   maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
   use_slots: True
+  pg_rewind:
+    username: postgres
+    password: zalando
   pg_hba:
   - host all all 0.0.0.0/0 md5
   - hostssl all all 0.0.0.0/0 md5
@@ -42,6 +45,7 @@ postgresql:
     password: rep-pass
     network: 127.0.0.1/32
   superuser:
+    user: postgres
     password: zalando
   admin:
     username: admin
@@ -62,3 +66,4 @@ postgresql:
     archive_timeout: 1800s
     max_replication_slots: 5
     hot_standby: "on"
+    wal_log_hints: "on"

From c8108f221e1f78162715cd4f70776e79b0365b38 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Thu, 24 Sep 2015 11:34:28 +0200
Subject: [PATCH 02/16] Check the exit code of the postgres start when
 determining whether to run the on_role_change callback.

---
 patroni/postgresql.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index b69099a9..51c3db13 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -328,7 +328,7 @@ recovery_target_timeline = 'latest'
                 # pg_rewind removes recovery.conf, we have to reinstate it.
                 if ret:
                     self.write_recovery_conf(leader)
-                    self.start()
+                    ret = self.start()
                 else:
                     self.remove_data_directory()
                     logger.error("unable to rewind the former leader")

From 027bcd39cede37524f788d0688bd3966c8e2259c Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Thu, 24 Sep 2015 12:46:36 +0200
Subject: [PATCH 03/16] Move pg_rewind call into a separate sub.

Add a Postgresql method to call pg_rewind.
Improve the test coverage.
---
 patroni/postgresql.py    | 38 ++++++++++++++++++++++----------------
 tests/test_postgresql.py | 23 ++++++++++++++++++++++-
 2 files changed, 44 insertions(+), 17 deletions(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 51c3db13..5d0e27da 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -47,7 +47,7 @@ class Postgresql:
         self.replication = config['replication']
         self.superuser = config['superuser']
         self.admin = config['admin']
-        self.pg_rewind = config.get('pg_rewind', {})
+        self._pg_rewind = config.get('pg_rewind', {})
         self.callback = config.get('callbacks', {})
         self.use_slots = config.get('use_slots', True)
         self.schedule_load_slots = self.use_slots
@@ -70,16 +70,19 @@ class Postgresql:
         self._cursor_holder = None
         self.members = []  # list of already existing replication slots
         self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException)
+        self.init_pg_rewind()
+
+    def init_pg_rewind(self):
         try:
-            self._pg_rewind_present = ('username' in self.pg_rewind and
+            self._pg_rewind_present = ('username' in self._pg_rewind and
                                        ('wal_log_hints' in self.config['parameters'] or
                                         'data_checksums' in self.config['parameters']) and
                                        os.system("pg_rewind --version >/dev/null 2>&1") == 0)
             if self._pg_rewind_present:
-                self.pg_rewind['user'] = self.pg_rewind['username']
+                self._pg_rewind['user'] = self._pg_rewind['username']
         except:
             self._pg_rewind_present = False
-        if self.pg_rewind and not self._pg_rewind_present:
+        if self._pg_rewind and not self._pg_rewind_present:
             logger.warning("pg_rewind support is disabled")
 
     def get_local_address(self):
@@ -310,6 +313,18 @@ recovery_target_timeline = 'latest'
                 for name, value in self.config.get('recovery_conf', {}).items():
                     f.write("{} = '{}'\n".format(name, value))
 
+    def pg_rewind(self, leader):
+        pc = self.primary_conninfo(leader.conn_url, self._pg_rewind) + ' dbname=postgres'
+        logger.info("running pg_rewind from {}".format(pc))
+        pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
+        try:
+            ret = (subprocess.call(pg_rewind) == 0)
+        except:
+            ret = False
+        if ret:
+            self.write_recovery_conf(leader)
+        return ret
+
     def follow_the_leader(self, leader):
         if not self.check_recovery_conf(leader):
             self.write_recovery_conf(leader)
@@ -317,20 +332,11 @@ recovery_target_timeline = 'latest'
 
             if leader and change_role and self._pg_rewind_present:
                 self.stop()
-                pc = self.primary_conninfo(leader.conn_url,
-                                           self.pg_rewind) + ' dbname=postgres'
-                logger.info("running pg_rewind from {}".format(pc))
-                pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
-                try:
-                    ret = (subprocess.call(pg_rewind) == 0)
-                except:
-                    ret = False
-                # pg_rewind removes recovery.conf, we have to reinstate it.
-                if ret:
-                    self.write_recovery_conf(leader)
+                if self.pg_rewind(leader):
                     ret = self.start()
                 else:
-                    self.remove_data_directory()
+                    ret = False
+                    self.move_data_directory()
                     logger.error("unable to rewind the former leader")
             else:
                 ret = self.restart()
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index eaf84943..35019780 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -9,6 +9,7 @@ from patroni.exceptions import PostgresConnectionException
 from patroni.postgresql import Postgresql
 from patroni.utils import RetryFailedError
 from test_ha import false
+import subprocess
 
 
 class MockCursor:
@@ -132,12 +133,32 @@ class TestPostgresql(unittest.TestCase):
     def test_sync_from_leader(self):
         self.assertTrue(self.p.sync_from_leader(self.leader))
 
-    def test_follow_the_leader(self):
+    @patch('os.system', side_effect=Exception("Test"))
+    def test_init_pg_rewind(self, mock_system):
+        self.p.init_pg_rewind()
+        # prepare parameters for pg_rewind
+        self.p._pg_rewind = {'username': 'foo'}
+        self.p.config['parameters']['data_checksums'] = 1
+        os.system = mock_system
+        self.p.init_pg_rewind()
+
+    @patch('subprocess.call', side_effect=Exception("Test"))
+    def test_pg_rewind(self, mock_call):
+        self.assertTrue(self.p.pg_rewind(self.leader))
+        self.p
+        subprocess.call = mock_call
+        self.assertFalse(self.p.pg_rewind(self.leader))
+
+    @patch('patroni.postgresql.Postgresql.pg_rewind', return_value=False)
+    def test_follow_the_leader(self, mock_pg_rewind):
         self.p.demote(self.leader)
         self.p.follow_the_leader(None)
+        self.p._pg_rewind_present = True
         self.p.demote(self.leader)
         self.p.follow_the_leader(self.leader)
         self.p.follow_the_leader(Leader(-1, None, 28, self.other))
+        self.p.pg_rewind = mock_pg_rewind
+        self.p.follow_the_leader(self.leader)
 
     def test_create_replica(self):
         self.p.delete_trigger_file = Mock(side_effect=OSError())

From d6c8df45e149a21e883124e3c09f72f5bff3606b Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Fri, 25 Sep 2015 13:08:12 +0200
Subject: [PATCH 04/16] Write the pg_rewind password in pgpass instead of
 passing it in the command line.

---
 patroni/postgresql.py | 25 ++++++++++++++++---------
 1 file changed, 16 insertions(+), 9 deletions(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 5d0e27da..be82a6e4 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -135,14 +135,17 @@ class Postgresql:
     def delete_trigger_file(self):
         os.path.exists(self.trigger_file) and os.unlink(self.trigger_file)
 
+    def write_pgpass(self, record, append=False):
+        pgpass = 'pgpass'
+        with open(pgpass, 'w' if not append else 'a') as f:
+            os.fchmod(f.fileno(), 0o600)
+            f.write('{host}:{port}:*:{user}:{password}\n'.format(**record))
+        return pgpass
+
     def sync_from_leader(self, leader):
         r = parseurl(leader.conn_url)
 
-        pgpass = 'pgpass'
-        with open(pgpass, 'w') as f:
-            os.fchmod(f.fileno(), 0o600)
-            f.write('{host}:{port}:*:{user}:{password}\n'.format(**r))
-
+        pgpass = self.write_pgpass(r)
         env = os.environ.copy()
         env['PGPASSFILE'] = pgpass
         return self.create_replica(r, env) == 0
@@ -280,10 +283,8 @@ class Postgresql:
                 f.write(line + '\n')
 
     @staticmethod
-    def primary_conninfo(leader_url, replacement=None):
+    def primary_conninfo(leader_url):
         r = parseurl(leader_url)
-        if replacement is not None:
-            r.update(replacement)
         return 'user={user} password={password} host={host} port={port} sslmode=prefer sslcompression=1'.format(**r)
 
     def check_recovery_conf(self, leader):
@@ -313,8 +314,14 @@ recovery_target_timeline = 'latest'
                 for name, value in self.config.get('recovery_conf', {}).items():
                     f.write("{} = '{}'\n".format(name, value))
 
+    def prepare_pg_rewind_connection(self, leader_url, pg_rewind):
+        r = parseurl(leader_url)
+        r.update(pg_rewind)
+        self.write_pgpass(r, append=True)
+        return "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
+
     def pg_rewind(self, leader):
-        pc = self.primary_conninfo(leader.conn_url, self._pg_rewind) + ' dbname=postgres'
+        pc = self.prepare_pg_rewind_connection(leader.conn_url, self._pg_rewind)
         logger.info("running pg_rewind from {}".format(pc))
         pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
         try:

From d44a54628ad7191da16279f6ae1d85d95ceee8b2 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Fri, 25 Sep 2015 16:00:24 +0200
Subject: [PATCH 05/16] remove the data directory on an unsuccessfull rewind
 attempt.

---
 patroni/postgresql.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 23047026..a7270460 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -343,7 +343,7 @@ recovery_target_timeline = 'latest'
                     ret = self.start()
                 else:
                     ret = False
-                    self.move_data_directory()
+                    self.remove_data_directory()
                     logger.error("unable to rewind the former leader")
             else:
                 ret = self.restart()

From e39d3187324a0e8dc08061bf0874c2fe026db836 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Mon, 28 Sep 2015 12:04:06 +0200
Subject: [PATCH 06/16] Eliminate os.system call.

---
 patroni/postgresql.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index a7270460..7e3e9088 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -77,7 +77,10 @@ class Postgresql:
             self._pg_rewind_present = ('username' in self._pg_rewind and
                                        ('wal_log_hints' in self.config['parameters'] or
                                         'data_checksums' in self.config['parameters']) and
-                                       os.system("pg_rewind --version >/dev/null 2>&1") == 0)
+                                       subprocess.call(['pg_rewind',
+                                                        '--version'],
+                                                       stdout=open(os.devnull, 'w'),
+                                                       stderr=subprocess.STDOUT) == 0)
             if self._pg_rewind_present:
                 self._pg_rewind['user'] = self._pg_rewind['username']
         except:

From a500781b6d9dca09c66d57af11920126203d3904 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Wed, 30 Sep 2015 16:32:56 +0200
Subject: [PATCH 07/16] Mock remove_data_directory in the pg_rewind test.

---
 tests/test_postgresql.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 3a14fd60..8551e02e 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -3,7 +3,7 @@ import psycopg2
 import shutil
 import unittest
 
-from mock import Mock, patch
+from mock import Mock, MagicMock, patch
 from patroni.dcs import Cluster, Leader, Member
 from patroni.exceptions import PostgresException, PostgresConnectionException
 from patroni.postgresql import Postgresql
@@ -150,6 +150,7 @@ class TestPostgresql(unittest.TestCase):
         self.assertFalse(self.p.pg_rewind(self.leader))
 
     @patch('patroni.postgresql.Postgresql.pg_rewind', return_value=False)
+    @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True))
     def test_follow_the_leader(self, mock_pg_rewind):
         self.p.demote(self.leader)
         self.p.follow_the_leader(None)

From b223319183d9a6813a6b0e37e7edd759daed94e0 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Wed, 30 Sep 2015 18:00:28 +0200
Subject: [PATCH 08/16] use the PATH to get the python interpreter path for the
 scripts.

---
 patroni/scripts/aws.py     | 2 +-
 patroni/scripts/restore.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/patroni/scripts/aws.py b/patroni/scripts/aws.py
index b172a8c3..a0622f15 100755
--- a/patroni/scripts/aws.py
+++ b/patroni/scripts/aws.py
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python
 
 import logging
 import requests
diff --git a/patroni/scripts/restore.py b/patroni/scripts/restore.py
index 4ac091f3..6b20e3e8 100755
--- a/patroni/scripts/restore.py
+++ b/patroni/scripts/restore.py
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python
 # arguments are:
 #   - cluster scope
 #   - cluster role

From ea910a89878910939dd4271d19fd87dc00dbbdc6 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Wed, 30 Sep 2015 18:02:07 +0200
Subject: [PATCH 09/16] Make sure pgpass file name is also passed in the
 PGPASSFILE environment variable.

---
 patroni/postgresql.py | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 7e3e9088..4c0da705 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -143,14 +143,14 @@ class Postgresql:
         with open(pgpass, 'w' if not append else 'a') as f:
             os.fchmod(f.fileno(), 0o600)
             f.write('{host}:{port}:*:{user}:{password}\n'.format(**record))
-        return pgpass
+        env = os.environ.copy()
+        env['PGPASSFILE'] = pgpass
+        return env
 
     def sync_from_leader(self, leader):
         r = parseurl(leader.conn_url)
 
-        pgpass = self.write_pgpass(r)
-        env = os.environ.copy()
-        env['PGPASSFILE'] = pgpass
+        env = self.write_pgpass(r)
         return self.create_replica(r, env) == 0
 
     @staticmethod
@@ -320,15 +320,15 @@ recovery_target_timeline = 'latest'
     def prepare_pg_rewind_connection(self, leader_url, pg_rewind):
         r = parseurl(leader_url)
         r.update(pg_rewind)
-        self.write_pgpass(r, append=True)
-        return "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
+        env = self.write_pgpass(r, append=True)
+        return (env, "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r))
 
     def pg_rewind(self, leader):
-        pc = self.prepare_pg_rewind_connection(leader.conn_url, self._pg_rewind)
+        env, pc = self.prepare_pg_rewind_connection(leader.conn_url, self._pg_rewind)
         logger.info("running pg_rewind from {}".format(pc))
         pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
         try:
-            ret = (subprocess.call(pg_rewind) == 0)
+            ret = (subprocess.call(pg_rewind, env=env) == 0)
         except:
             ret = False
         if ret:

From bad37a5a212ebaf81a0d282fa1133a51785ce183 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Fri, 2 Oct 2015 10:57:49 +0200
Subject: [PATCH 10/16] Always check that cluster is configured correctly right
 before running pg_rewind.

---
 patroni/postgresql.py | 25 +++++++++++++++----------
 1 file changed, 15 insertions(+), 10 deletions(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 4c0da705..4fd0e13d 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -75,8 +75,6 @@ class Postgresql:
     def init_pg_rewind(self):
         try:
             self._pg_rewind_present = ('username' in self._pg_rewind and
-                                       ('wal_log_hints' in self.config['parameters'] or
-                                        'data_checksums' in self.config['parameters']) and
                                        subprocess.call(['pg_rewind',
                                                         '--version'],
                                                        stdout=open(os.devnull, 'w'),
@@ -317,14 +315,12 @@ recovery_target_timeline = 'latest'
                 for name, value in self.config.get('recovery_conf', {}).items():
                     f.write("{} = '{}'\n".format(name, value))
 
-    def prepare_pg_rewind_connection(self, leader_url, pg_rewind):
-        r = parseurl(leader_url)
-        r.update(pg_rewind)
-        env = self.write_pgpass(r, append=True)
-        return (env, "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r))
-
     def pg_rewind(self, leader):
-        env, pc = self.prepare_pg_rewind_connection(leader.conn_url, self._pg_rewind)
+        # prepare pg_rewind connection
+        r = parseurl(leader.conn_url)
+        r.update(self._pg_rewind)
+        env = self.write_pgpass(r, append=True)
+        pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
         logger.info("running pg_rewind from {}".format(pc))
         pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
         try:
@@ -335,12 +331,21 @@ recovery_target_timeline = 'latest'
             self.write_recovery_conf(leader)
         return ret
 
+    def pg_rewind_verify_cluster(self):
+        """ check that pg_rewind can be used with the cluster """
+        try:
+            return self.query("""SELECT bool_or(setting::boolean)
+                                 FROM pg_settings
+                                 WHERE name IN ( 'data_checksums', 'wal_log_hints')""").fetchone()[0]
+        except:
+            return False
+
     def follow_the_leader(self, leader):
         if not self.check_recovery_conf(leader):
             self.write_recovery_conf(leader)
             change_role = self.role == 'master'
 
-            if leader and change_role and self._pg_rewind_present:
+            if leader and change_role and self._pg_rewind_present and self.pg_rewind_verify_cluster():
                 self.stop()
                 if self.pg_rewind(leader):
                     ret = self.start()

From b629e0852f8f9d5a5072b24268e06f025c45a9d5 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Mon, 12 Oct 2015 08:34:08 +0200
Subject: [PATCH 11/16] Call pg_rewind in case of the master's unclean
 shutdown.

If patroni detects the former master was killed, it runs
it first in a single-user mode and then shuts down normally,
to make sure pg_rewind will see a normal shut down status
in pg_controldata.

Add a flag need_rewind, since the point where it is detected
that rewind might be necessary is moved out the code that
runs rewind.
---
 patroni/ha.py            |  11 ++-
 patroni/postgresql.py    | 155 +++++++++++++++++++++++++++--------
 tests/test_postgresql.py | 170 +++++++++++++++++++++++++++++++++++----
 3 files changed, 286 insertions(+), 50 deletions(-)

diff --git a/patroni/ha.py b/patroni/ha.py
index 0d95d372..41a2a56a 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -66,8 +66,15 @@ class Ha:
         if self.state_handler.is_healthy():
             return False
         has_lock = self.has_lock()
-        self.state_handler.write_recovery_conf(None if has_lock else self.cluster.leader)
-        self.state_handler.start()
+
+        # try to see if we are the former master that crashed. If so - we likely need to run pg_rewind
+        # in order to join the former standby being promoted.
+        pg_controldata = self.state_handler.controldata()
+        if not has_lock and pg_controldata.get('Database cluster state', '') == 'in production':  # crashed master
+            self.state_handler.require_rewind()
+
+        # XXX: should we call ha.follow_the_leader here instead?
+        ret = self.state_handler.follow_the_leader(None if has_lock else self.cluster.leader, recovery=True)
         if has_lock:
             logger.info('started as readonly because i had the session lock')
             self.load_cluster_from_dcs()
diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 4fd0e13d..59ee4936 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -47,7 +47,7 @@ class Postgresql:
         self.replication = config['replication']
         self.superuser = config['superuser']
         self.admin = config['admin']
-        self._pg_rewind = config.get('pg_rewind', {})
+        self.pg_rewind = config.get('pg_rewind', {})
         self.callback = config.get('callbacks', {})
         self.use_slots = config.get('use_slots', True)
         self.schedule_load_slots = self.use_slots
@@ -68,23 +68,36 @@ class Postgresql:
 
         self._connection = None
         self._cursor_holder = None
+        self._need_rewind = False
         self.members = []  # list of already existing replication slots
         self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException)
-        self.init_pg_rewind()
 
-    def init_pg_rewind(self):
+    @property
+    def can_rewind(self):
+        """ check if pg_rewind executable is there and that pg_controldata indicates
+            we have either wal_log_hints or checksums turned on
+        """
+        # low-hanging fruit: check if pg_rewind configuration is there
+        if not self.pg_rewind or\
+           not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')):
+            return False
+
+        cmd = ['pg_rewind', '--help']
         try:
-            self._pg_rewind_present = ('username' in self._pg_rewind and
-                                       subprocess.call(['pg_rewind',
-                                                        '--version'],
-                                                       stdout=open(os.devnull, 'w'),
-                                                       stderr=subprocess.STDOUT) == 0)
-            if self._pg_rewind_present:
-                self._pg_rewind['user'] = self._pg_rewind['username']
-        except:
-            self._pg_rewind_present = False
-        if self._pg_rewind and not self._pg_rewind_present:
-            logger.warning("pg_rewind support is disabled")
+            ret = subprocess.call(cmd, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
+            if ret != 0:  # pg_rewind is not there, close up the shop and go home
+                return False
+        except OSError:
+            return False
+        # check if the cluster's configuration permits pg_rewind
+        data = self.controldata()
+        if data:
+            return data.get('wal_log_hints setting', 'off') == 'on' or\
+                data.get('Data page checksum version', '0') != '0'
+        return False
+
+    def require_rewind(self):
+        self._need_rewind = True
 
     def get_local_address(self):
         listen_addresses = self.listen_addresses.split(',')
@@ -209,6 +222,8 @@ class Postgresql:
         return ret
 
     def stop(self, mode='fast', block_callbacks=False):
+        if not self.is_running():
+            return True
         if block_callbacks:
             try:
                 self.query('SET statement_timeout TO 0')
@@ -315,10 +330,11 @@ recovery_target_timeline = 'latest'
                 for name, value in self.config.get('recovery_conf', {}).items():
                     f.write("{} = '{}'\n".format(name, value))
 
-    def pg_rewind(self, leader):
+    def rewind(self, leader):
         # prepare pg_rewind connection
         r = parseurl(leader.conn_url)
-        r.update(self._pg_rewind)
+        r.update(self.pg_rewind)
+        r['user'] = r['username']
         env = self.write_pgpass(r, append=True)
         pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
         logger.info("running pg_rewind from {}".format(pc))
@@ -331,30 +347,99 @@ recovery_target_timeline = 'latest'
             self.write_recovery_conf(leader)
         return ret
 
-    def pg_rewind_verify_cluster(self):
-        """ check that pg_rewind can be used with the cluster """
+    def controldata(self):
+        """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """
+        result = None
         try:
-            return self.query("""SELECT bool_or(setting::boolean)
-                                 FROM pg_settings
-                                 WHERE name IN ( 'data_checksums', 'wal_log_hints')""").fetchone()[0]
-        except:
-            return False
+            data = subprocess.check_output(['pg_controldata', self.data_dir])
+            if data:
+                data = data.splitlines()
+                result = {l.split(':')[0]: l.split(':')[1].strip() for l in data if l}
+        except subprocess.CalledProcessError:
+            logger.exception("Error when calling pg_controldata")
+        finally:
+            return result
 
-    def follow_the_leader(self, leader):
-        if not self.check_recovery_conf(leader):
+    def read_postmaster_opts(self):
+        """ returns the list of option names/values from postgres.opts, Empty dict if read failed or no file """
+        result = {}
+        try:
+            with open(os.path.join(self.data_dir, "postmaster.opts")) as f:
+                data = f.read()
+                opts = [opt.strip('"\n') for opt in data.split(' "')]
+                for opt in opts:
+                    if '=' in opt and opt.startswith('--'):
+                        name, val = opt.split('=', 1)
+                        name = name.strip('-')
+                        result[name] = val
+        except IOError:
+            logger.exception('Error when reading postmaster.opts')
+        finally:
+            return result
+
+    def single_user_mode(self, command=None, options={}):
+        """ run a given command in a single-user mode. If the command is empty - then just start and stop """
+        cmd = ['postgres', '--single', '-D', self.data_dir]
+        for opt in sorted(options):
+            cmd.extend(['-c', '{0}={1}'.format(opt, options[opt])])
+        # need a database name to connect
+        cmd.append('postgres')
+        p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
+        if p:
+            command and p.communicate('{}\n'.format(command))
+            p.stdin.close()
+            return p.wait()
+        return 1
+
+    def cleanup_archive_status(self):
+        status_dir = os.path.join(self.data_dir, 'pg_xlog', 'archive_status')
+        if os.path.isdir(status_dir):
+            for f in os.listdir(status_dir):
+                path = os.path.join(status_dir, f)
+                try:
+                    if os.path.isfile(path):
+                        os.remove(path)
+                    elif os.path.islink(path):  # should not happen, but just in case
+                        os.unlink(path)
+                except:
+                    logger.exception("Unable to remove {}".format(path))
+
+    def follow_the_leader(self, leader, recovery=False):
+        if not self.check_recovery_conf(leader) or recovery:
+            change_role = (self.role == 'master')
+
+            self._need_rewind = (self._need_rewind or change_role) and self.can_rewind
+            if self._need_rewind:
+                logger.info("set the rewind flag after demote")
             self.write_recovery_conf(leader)
-            change_role = self.role == 'master'
-
-            if leader and change_role and self._pg_rewind_present and self.pg_rewind_verify_cluster():
-                self.stop()
-                if self.pg_rewind(leader):
+            if not leader or not self._need_rewind:  # do not rewind until the leader becomes available
+                ret = self.restart()
+            else:  # we have a leader and need to rewind
+                if self.is_running():
+                    self.stop()
+                # at present, pg_rewind only runs when the cluster is shut down cleanly
+                # and not shutdown in recovery. We have to remove the recovery.conf if present
+                # and start/shutdown in a single user mode to emulate this.
+                # XXX: if recovery.conf is linked, it will be written anew as a normal file.
+                if os.path.isfile(self.recovery_conf):
+                    os.remove(self.recovery_conf)
+                else:
+                    os.unlink(self.recovery_conf)
+                # Archived segments might be useful to pg_rewind,
+                # clean the flags that tell we should remove them.
+                self.cleanup_archive_status()
+                # Start in a single user mode and stop to produce a clean shutdown
+                opts = self.read_postmaster_opts()
+                opts['archive_mode'] = 'on'
+                opts['archive_command'] = 'false'
+                self.single_user_mode(options=opts)
+                if self.rewind(leader):
                     ret = self.start()
                 else:
-                    ret = False
+                    logger.error("unable to rewind the former master")
                     self.remove_data_directory()
-                    logger.error("unable to rewind the former leader")
-            else:
-                ret = self.restart()
+                    ret = True
+                self._need_rewind = False
             change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE)
 
     def save_configuration_files(self):
@@ -379,6 +464,8 @@ recovery_target_timeline = 'latest'
         ret = subprocess.call(self._pg_ctl + ['promote']) == 0
         if ret:
             self._role = 'master'
+            logger.info("cleared rewind flag after becoming the leader")
+            self._need_rewind = False
             self.call_nowait(ACTION_ON_ROLE_CHANGE)
         return ret
 
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 8551e02e..96b1cbe4 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -1,8 +1,15 @@
+import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
 import psycopg2
 import shutil
 import unittest
 
+from sys import version_info
+if version_info.major == 2:
+    import __builtin__ as builtins
+else:
+    import builtins
+
 from mock import Mock, MagicMock, patch
 from patroni.dcs import Cluster, Leader, Member
 from patroni.exceptions import PostgresException, PostgresConnectionException
@@ -81,6 +88,68 @@ class MockConnect(Mock):
         return MockCursor(self)
 
 
+def pg_controldata_string(*args, **kwargs):
+    return """
+pg_control version number:            942
+Catalog version number:               201509161
+Database system identifier:           6200971513092291716
+Database cluster state:               shut down in recovery
+pg_control last modified:             Fri Oct  2 10:57:06 2015
+Latest checkpoint location:           0/30000C8
+Prior checkpoint location:            0/2000060
+Latest checkpoint's REDO location:    0/3000090
+Latest checkpoint's REDO WAL file:    000000020000000000000003
+Latest checkpoint's TimeLineID:       2
+Latest checkpoint's PrevTimeLineID:   2
+Latest checkpoint's full_page_writes: on
+Latest checkpoint's NextXID:          0/943
+Latest checkpoint's NextOID:          24576
+Latest checkpoint's NextMultiXactId:  1
+Latest checkpoint's NextMultiOffset:  0
+Latest checkpoint's oldestXID:        931
+Latest checkpoint's oldestXID's DB:   1
+Latest checkpoint's oldestActiveXID:  943
+Latest checkpoint's oldestMultiXid:   1
+Latest checkpoint's oldestMulti's DB: 1
+Latest checkpoint's oldestCommitTs:   0
+Latest checkpoint's newestCommitTs:   0
+Time of latest checkpoint:            Fri Oct  2 10:56:54 2015
+Fake LSN counter for unlogged rels:   0/1
+Minimum recovery ending location:     0/30241F8
+Min recovery ending loc's timeline:   2
+Backup start location:                0/0
+Backup end location:                  0/0
+End-of-backup record required:        no
+wal_level setting:                    hot_standby
+wal_log_hints setting:                on
+max_connections setting:              100
+max_worker_processes setting:         8
+max_prepared_xacts setting:           0
+max_locks_per_xact setting:           64
+track_commit_timestamp setting:       off
+Maximum data alignment:               8
+Database block size:                  8192
+Blocks per segment of large relation: 131072
+WAL block size:                       8192
+Bytes per WAL segment:                16777216
+Maximum length of identifiers:        64
+Maximum columns in an index:          32
+Maximum size of a TOAST chunk:        1996
+Size of a large-object chunk:         2048
+Date/time type storage:               64-bit integers
+Float4 argument passing:              by value
+Float8 argument passing:              by value
+Data page checksum version:           0
+"""
+
+
+def postmaster_opts_string(*args, **kwargs):
+    return '/usr/local/pgsql/bin/postgres "-D" "data/postgresql0" "--listen_addresses=127.0.0.1" "--port=5432"'\
+        ' "--hot_standby=on" "--wal_keep_segments=8" "--wal_level=hot_standby" "--archive_command=mkdir -p ../wal_archive \n'\
+        '&& cp %p ../wal_archive/%f" "--wal_log_hints=on" "--max_wal_senders=5" "--archive_timeout=1800s" "--archive_mode=on"'\
+        ' "--max_replication_slots=5"\n'
+
+
 def psycopg2_connect(*args, **kwargs):
     return MockConnect()
 
@@ -96,6 +165,7 @@ class TestPostgresql(unittest.TestCase):
                              'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'],
                              'superuser': {'password': ''},
                              'admin': {'username': 'admin', 'password': 'admin'},
+                             'pg_rewind': {'username': 'admin', 'password': 'admin'},
                              'replication': {'username': 'replicator',
                                              'password': 'rep-pass',
                                              'network': '127.0.0.1/32'},
@@ -133,32 +203,22 @@ class TestPostgresql(unittest.TestCase):
     def test_sync_from_leader(self):
         self.assertTrue(self.p.sync_from_leader(self.leader))
 
-    @patch('os.system', side_effect=Exception("Test"))
-    def test_init_pg_rewind(self, mock_system):
-        self.p.init_pg_rewind()
-        # prepare parameters for pg_rewind
-        self.p._pg_rewind = {'username': 'foo'}
-        self.p.config['parameters']['data_checksums'] = 1
-        os.system = mock_system
-        self.p.init_pg_rewind()
-
     @patch('subprocess.call', side_effect=Exception("Test"))
     def test_pg_rewind(self, mock_call):
-        self.assertTrue(self.p.pg_rewind(self.leader))
+        self.assertTrue(self.p.rewind(self.leader))
         self.p
         subprocess.call = mock_call
-        self.assertFalse(self.p.pg_rewind(self.leader))
+        self.assertFalse(self.p.rewind(self.leader))
 
-    @patch('patroni.postgresql.Postgresql.pg_rewind', return_value=False)
+    @patch('patroni.postgresql.Postgresql.rewind', return_value=False)
     @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True))
     def test_follow_the_leader(self, mock_pg_rewind):
         self.p.demote(self.leader)
         self.p.follow_the_leader(None)
-        self.p._pg_rewind_present = True
         self.p.demote(self.leader)
         self.p.follow_the_leader(self.leader)
         self.p.follow_the_leader(Leader(-1, None, 28, self.other))
-        self.p.pg_rewind = mock_pg_rewind
+        self.p.rewind = mock_pg_rewind
         self.p.follow_the_leader(self.leader)
 
     def test_create_replica(self):
@@ -248,3 +308,85 @@ class TestPostgresql(unittest.TestCase):
         with patch('os.unlink', Mock(side_effect=Exception)):
             self.p.remove_data_directory()
         self.p.remove_data_directory()
+
+    @patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string))
+    @patch('subprocess.check_output', side_effect=subprocess.CalledProcessError)
+    @patch('subprocess.check_output', side_effect=Exception('Failed'))
+    def test_controldata(self, check_output_call_error, check_output_generic_exception):
+        data = self.p.controldata()
+        self.assertEquals(len(data), 50)
+        self.assertEquals(data['Database cluster state'], 'shut down in recovery')
+        self.assertEquals(data['wal_log_hints setting'], 'on')
+        self.assertEquals(int(data['Database block size']), 8192)
+
+        subprocess.check_output = check_output_call_error
+        data = self.p.controldata()
+        self.assertIsNone(data)
+
+        subprocess.check_output = check_output_generic_exception
+        self.assertRaises(Exception, self.p.controldata())
+
+    def test_read_postmaster_opts(self):
+        m = mock.mock_open(read_data=postmaster_opts_string())
+        with patch.object(builtins, 'open', m):
+            data = self.p.read_postmaster_opts()
+            self.assertEquals(data['wal_level'], 'hot_standby')
+            self.assertEquals(int(data['max_replication_slots']), 5)
+            self.assertEqual(data.get('D'), None)
+
+            m.side_effect = IOError("foo")
+            data = self.p.read_postmaster_opts()
+            self.assertEqual(data, dict())
+
+            m.side_effect = Exception("foo")
+            self.assertRaises(Exception, self.p.read_postmaster_opts())
+
+    @patch('subprocess.Popen')
+    @patch.object(builtins, 'open', MagicMock(return_value=42))
+    def test_single_user_mode(self, subprocess_popen_mock):
+        subprocess_popen_mock.return_value.wait.return_value = 0
+        self.assertEquals(self.p.single_user_mode(options=dict(archive_mode='on', archive_command='false')), 0)
+        subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir,
+                                                      '-c', 'archive_command=false', '-c', 'archive_mode=on',
+                                                       'postgres'], stdin=subprocess.PIPE,
+                                                      stdout=42,
+                                                      stderr=subprocess.STDOUT)
+        subprocess_popen_mock.reset_mock()
+        self.assertEquals(self.p.single_user_mode(command="CHECKPOINT"), 0)
+        subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir,
+                                                      'postgres'], stdin=subprocess.PIPE,
+                                                      stdout=42,
+                                                      stderr=subprocess.STDOUT)
+        subprocess_popen_mock.return_value = None
+        self.assertEquals(self.p.single_user_mode(), 1)
+
+    def fake_listdir(path):
+        if path.endswith(os.path.join('pg_xlog', 'archive_status')):
+            return ["a", "b", "c"]
+        return []
+
+    @patch('os.listdir', MagicMock(side_effect=fake_listdir))
+    @patch('os.path.isdir', MagicMock(return_value=True))
+    @patch('os.unlink', return_value=True)
+    @patch('os.remove', return_value=True)
+    @patch('os.path.islink', return_value=False)
+    @patch('os.path.isfile', return_value=True)
+    def test_cleanup_archive_status(self, mock_file, mock_link, mock_remove, mock_unlink):
+        ap = os.path.join(self.p.data_dir, 'pg_xlog', 'archive_status/')
+        self.p.cleanup_archive_status()
+        mock_remove.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')])
+        mock_unlink.assert_not_called()
+
+        mock_remove.reset_mock()
+        mock_file.return_value = False
+        mock_link.return_value = True
+        self.p.cleanup_archive_status()
+        mock_unlink.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')])
+        mock_remove.assert_not_called()
+
+        mock_unlink.reset_mock()
+        mock_remove.reset_mock()
+        mock_file.side_effect = Exception("foo")
+        self.p.cleanup_archive_status()
+        mock_unlink.assert_not_called()
+        mock_remove.assert_not_called()

From ce7169f61df109427ec723519d24773bddc1b6fb Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Mon, 12 Oct 2015 15:29:47 +0200
Subject: [PATCH 12/16] Add new tests ha and postgresql.

---
 tests/test_ha.py         | 1 +
 tests/test_postgresql.py | 9 +++++++--
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/tests/test_ha.py b/tests/test_ha.py
index c55cbd90..38bb07a6 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -117,6 +117,7 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.run_cycle(), 'started as a secondary')
 
     def test_recover_replica_failed(self):
+        self.p.controldata = lambda: {'Database cluster state': 'in production'}
         self.p.is_healthy = false
         self.p.follow_the_leader = false
         self.assertEquals(self.ha.run_cycle(), 'failed to start postgres')
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 0be4317e..060302c7 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -10,7 +10,7 @@ if version_info.major == 2:
 else:
     import builtins
 
-from mock import Mock, MagicMock, patch
+from mock import Mock, MagicMock, PropertyMock, patch
 from patroni.dcs import Cluster, Leader, Member
 from patroni.exceptions import PostgresException, PostgresConnectionException
 from patroni.postgresql import Postgresql
@@ -216,7 +216,6 @@ class TestPostgresql(unittest.TestCase):
     @patch('subprocess.call', side_effect=Exception("Test"))
     def test_pg_rewind(self, mock_call):
         self.assertTrue(self.p.rewind(self.leader))
-        self.p
         subprocess.call = mock_call
         self.assertFalse(self.p.rewind(self.leader))
 
@@ -230,6 +229,12 @@ class TestPostgresql(unittest.TestCase):
         self.p.follow_the_leader(Leader(-1, 28, self.other))
         self.p.rewind = mock_pg_rewind
         self.p.follow_the_leader(self.leader)
+        self.p.require_rewind()
+        with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
+            self.p.rewind.return_value = True
+            self.p.follow_the_leader(self.leader, recovery=True)
+            self.p.rewind.return_value = False
+            self.p.follow_the_leader(self.leader, recovery=True)
 
     def test_create_replica(self):
         self.p.delete_trigger_file = Mock(side_effect=OSError())

From d7988384d37a533bba8f945e60103d72a6de37a3 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Mon, 12 Oct 2015 16:24:02 +0200
Subject: [PATCH 13/16] Address the code review by Alex Kukushkin:

- check the link before checking the file when deciding to remove it,
  as isfile follows symlinks and, therefore, may return True on them.
- Remove append mode from write_pgpass, as it is always written anew
  before it is used.
- make pg_controldata return an empty hash in case of an error, and
  check for the empty value return by this function before using it.
 some other minior fixed and test updates.
---
 patroni/ha.py            |  3 ++-
 patroni/postgresql.py    | 22 +++++++++++-----------
 tests/test_postgresql.py |  5 ++++-
 3 files changed, 17 insertions(+), 13 deletions(-)

diff --git a/patroni/ha.py b/patroni/ha.py
index 3bb75b78..0019253d 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -96,7 +96,8 @@ class Ha:
         # try to see if we are the former master that crashed. If so - we likely need to run pg_rewind
         # in order to join the former standby being promoted.
         pg_controldata = self.state_handler.controldata()
-        if not has_lock and pg_controldata.get('Database cluster state', '') == 'in production':  # crashed master
+        if not has_lock and pg_controldata and\
+                pg_controldata.get('Database cluster state', '') == 'in production':  # crashed master
             self.state_handler.require_rewind()
 
         # XXX: follow the leader calls stop, which might take quite some time.
diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 2051d819..3ab8b345 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -164,9 +164,9 @@ class Postgresql:
     def delete_trigger_file(self):
         os.path.exists(self.trigger_file) and os.unlink(self.trigger_file)
 
-    def write_pgpass(self, record, append=False):
+    def write_pgpass(self, record):
         pgpass = 'pgpass'
-        with open(pgpass, 'w' if not append else 'a') as f:
+        with open(pgpass, 'w') as f:
             os.fchmod(f.fileno(), 0o600)
             f.write('{host}:{port}:*:{user}:{password}\n'.format(**record))
         env = os.environ.copy()
@@ -363,7 +363,7 @@ recovery_target_timeline = 'latest'
         r = parseurl(leader.conn_url)
         r.update(self.pg_rewind)
         r['user'] = r['username']
-        env = self.write_pgpass(r, append=True)
+        env = self.write_pgpass(r)
         pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
         logger.info("running pg_rewind from {}".format(pc))
         pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
@@ -377,7 +377,7 @@ recovery_target_timeline = 'latest'
 
     def controldata(self):
         """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """
-        result = None
+        result = {}
         try:
             data = subprocess.check_output(['pg_controldata', self.data_dir])
             if data:
@@ -425,10 +425,10 @@ recovery_target_timeline = 'latest'
             for f in os.listdir(status_dir):
                 path = os.path.join(status_dir, f)
                 try:
-                    if os.path.isfile(path):
-                        os.remove(path)
-                    elif os.path.islink(path):  # should not happen, but just in case
+                    if os.path.islink(path):
                         os.unlink(path)
+                    elif os.path.isfile(path):
+                        os.remove(path)
                 except:
                     logger.exception("Unable to remove {}".format(path))
 
@@ -449,10 +449,10 @@ recovery_target_timeline = 'latest'
                 # and not shutdown in recovery. We have to remove the recovery.conf if present
                 # and start/shutdown in a single user mode to emulate this.
                 # XXX: if recovery.conf is linked, it will be written anew as a normal file.
-                if os.path.isfile(self.recovery_conf):
-                    os.remove(self.recovery_conf)
-                else:
+                if os.path.islink(self.recovery_conf):
                     os.unlink(self.recovery_conf)
+                else:
+                    os.remove(self.recovery_conf)
                 # Archived segments might be useful to pg_rewind,
                 # clean the flags that tell we should remove them.
                 self.cleanup_archive_status()
@@ -494,7 +494,7 @@ recovery_target_timeline = 'latest'
             return True
         ret = subprocess.call(self._pg_ctl + ['promote']) == 0
         if ret:
-            self._role = 'master'
+            self.set_role('master')
             logger.info("cleared rewind flag after becoming the leader")
             self._need_rewind = False
             self.call_nowait(ACTION_ON_ROLE_CHANGE)
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 060302c7..53f41b98 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -337,7 +337,7 @@ class TestPostgresql(unittest.TestCase):
 
         subprocess.check_output = check_output_call_error
         data = self.p.controldata()
-        self.assertIsNone(data)
+        self.assertEquals(data, dict())
 
         subprocess.check_output = check_output_generic_exception
         self.assertRaises(Exception, self.p.controldata())
@@ -394,6 +394,7 @@ class TestPostgresql(unittest.TestCase):
         mock_unlink.assert_not_called()
 
         mock_remove.reset_mock()
+
         mock_file.return_value = False
         mock_link.return_value = True
         self.p.cleanup_archive_status()
@@ -402,7 +403,9 @@ class TestPostgresql(unittest.TestCase):
 
         mock_unlink.reset_mock()
         mock_remove.reset_mock()
+
         mock_file.side_effect = Exception("foo")
+        mock_link.side_effect = Exception("foo")
         self.p.cleanup_archive_status()
         mock_unlink.assert_not_called()
         mock_remove.assert_not_called()

From 46f4788c28c9e0cd3052ba83273f0b704f5c1224 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Mon, 12 Oct 2015 17:06:13 +0200
Subject: [PATCH 14/16] Do not try to run postgres -D during unit tests.

---
 tests/test_postgresql.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 53f41b98..5677c638 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -221,6 +221,7 @@ class TestPostgresql(unittest.TestCase):
 
     @patch('patroni.postgresql.Postgresql.rewind', return_value=False)
     @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True))
+    @patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1))
     def test_follow_the_leader(self, mock_pg_rewind):
         self.p.demote()
         self.p.follow_the_leader(None)

From 94aa6873f4b4e5eae460f6897d335559c07dc0f8 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Tue, 13 Oct 2015 08:19:44 +0200
Subject: [PATCH 15/16] Add more tests for the new postgresql methods.

---
 tests/test_postgresql.py | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 5677c638..2f1c75ac 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -237,6 +237,20 @@ class TestPostgresql(unittest.TestCase):
             self.p.rewind.return_value = False
             self.p.follow_the_leader(self.leader, recovery=True)
 
+    def test_can_rewind(self):
+        tmp = self.p.pg_rewind
+        self.p.pg_rewind = None
+        self.assertFalse(self.p.can_rewind)
+        self.p.pg_rewind = tmp
+        with mock.patch('subprocess.call', MagicMock(return_value=1)):
+            self.assertFalse(self.p.can_rewind)
+        with mock.patch('subprocess.call', side_effect=OSError("foo")):
+            self.assertFalse(self.p.can_rewind)
+        tmp = self.p.controldata()
+        self.p.controldata = lambda: {'wal_log_hints setting': 'on'}
+        self.assertTrue(self.p.can_rewind)
+        self.p.controldata = tmp
+
     def test_create_replica(self):
         self.p.delete_trigger_file = Mock(side_effect=OSError())
         self.assertEquals(self.p.create_replica({'host': '', 'port': '', 'user': ''}, ''), 1)

From 101082fa3b0ef02a4369b1c373f99a2e3ad9b6f0 Mon Sep 17 00:00:00 2001
From: Oleksii Kliukin <oleksii.kliukin@zalando.de>
Date: Tue, 13 Oct 2015 09:08:27 +0200
Subject: [PATCH 16/16] more tests.

---
 patroni/postgresql.py    | 5 +----
 tests/test_postgresql.py | 9 +++++++--
 2 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 3ab8b345..fc7956e9 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -340,10 +340,7 @@ class Postgresql:
         with open(self.recovery_conf, 'r') as f:
             for line in f:
                 if line.startswith('primary_conninfo'):
-                    if not pattern:
-                        return False
-                    return pattern in line
-
+                    return pattern and (pattern in line)
         return not pattern
 
     def write_recovery_conf(self, leader):
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 2f1c75ac..e5e2dfd4 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -10,7 +10,7 @@ if version_info.major == 2:
 else:
     import builtins
 
-from mock import Mock, MagicMock, PropertyMock, patch
+from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 from patroni.dcs import Cluster, Leader, Member
 from patroni.exceptions import PostgresException, PostgresConnectionException
 from patroni.postgresql import Postgresql
@@ -231,6 +231,11 @@ class TestPostgresql(unittest.TestCase):
         self.p.rewind = mock_pg_rewind
         self.p.follow_the_leader(self.leader)
         self.p.require_rewind()
+        with mock.patch('os.path.islink', MagicMock(return_value=True)):
+            with mock.patch('os.unlink', MagicMock(return_value=True)):
+                with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
+                    self.p.follow_the_leader(self.leader, recovery=True)
+        self.p.require_rewind()
         with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
             self.p.rewind.return_value = True
             self.p.follow_the_leader(self.leader, recovery=True)
@@ -358,7 +363,7 @@ class TestPostgresql(unittest.TestCase):
         self.assertRaises(Exception, self.p.controldata())
 
     def test_read_postmaster_opts(self):
-        m = mock.mock_open(read_data=postmaster_opts_string())
+        m = mock_open(read_data=postmaster_opts_string())
         with patch.object(builtins, 'open', m):
             data = self.p.read_postmaster_opts()
             self.assertEquals(data['wal_level'], 'hot_standby')