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

@@ -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()

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
 

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()