Fix pg_rewind behaviour (#524)

When Patroni does calculation whether it should run pg_rewind or not, it relies on pg_controldata output or gets necessary information from replication connection.
On some cases (when for example postgres running as a master was killed), we can't use pg_controldata output immediately, but trying to start postgres. Such start could fail with the following errror:
```
LOG,00000,"ending log output to stderr",,"Future log output will go to log destination ""csvlog"".",,,,,,,""
LOG,00000,"database system was interrupted; last known up at 2017-09-16 22:35:22 UTC",,,,,,,,,""
LOG,00000,"restored log file ""00000006.history"" from archive",,,,,,,,,""
LOG,00000,"entering standby mode",,,,,,,,,"" 2017-09-18 08:00:39.433 UTC,,,57,,59bf7d26.39,4,,2017-09-18 08:00:38 UTC,,0,LOG,00000,"restored log file ""00000006.history"" from archive",,,,,,,,,""
FATAL,XX000,"requested timeline 6 is not a child of this server's history","Latest checkpoint is at 29/1A000178 on timeline 5, but in the history of the requested timeline, the server forked off from that timeline at 29/1A000140.",,,,,,,,""
LOG,00000,"startup process (PID 57) exited with exit code 1",,,,,,,,,""
LOG,00000,"aborting startup due to startup process failure",,,,,,,,,""
LOG,00000,"database system is shut down",,,,,,,,,""
```
In this case controldata will still have `Database cluster state: in production`
All further attempts to start postgres will fail. Such situation could be fixed only if we start not in recovery. For safety we will do it in a single user mode.

The second problems is: if postgres was running as master, but later we started it and stopped, than pg_controldata will report:
```
Database cluster state:               shut down in recovery
Minimum recovery ending location:     0/0
Min recovery ending loc's timeline:   0
```

And this info can't be used for calculations. In this case we should use
`Latest checkpoint location` and `Latest checkpoint's TimeLineID`

patroni/ha.py

@@ -207,6 +207,16 @@ class Ha(object):
             msg = "starting as a secondary"
             node_to_follow = self._get_node_to_follow(self.cluster)
 
+        # once we already tried to start postgres but failed, single user mode is a rescue in this case
+        if self.recovering and not self.state_handler.rewind_executed and self.state_handler.can_rewind:
+            data = self.state_handler.controldata()
+            if data.get('Database cluster state') not in ('shut down', 'shut down in recovery'):
+                self.recovering = False
+                msg = 'fixing cluster state in a single user mode'
+                self._async_executor.schedule(msg)
+                self._async_executor.run_async(self.state_handler.fix_cluster_state)
+                return msg
+
         self.recovering = True
 
         self._async_executor.schedule('restarting after failure')

patroni/postgresql.py

@@ -1261,12 +1261,14 @@ class Postgresql(object):
         else:  # otherwise analyze pg_controldata output
             data = self.controldata()
             try:
+                if data.get('Database cluster state') == 'shut down in recovery':
+                    lsn = data.get('Minimum recovery ending location')
+                    timeline = int(data.get("Min recovery ending loc's timeline"))
+                    if lsn == '0/0' or timeline == 0:  # it was a master when it crashed
+                        data['Database cluster state'] = 'shut down'
                 if data.get('Database cluster state') == 'shut down':
                     lsn = data.get('Latest checkpoint location')
                     timeline = int(data.get("Latest checkpoint's TimeLineID"))
-                elif data.get('Database cluster state') == 'shut down in recovery':
-                    lsn = data.get('Minimum recovery ending location')
-                    timeline = int(data.get("Min recovery ending loc's timeline"))
             except (TypeError, ValueError):
                 logger.exception('Failed to get local timeline and lsn from pg_controldata output')
         logger.info('Local timeline=%s lsn=%s', timeline, lsn)
@@ -1744,3 +1746,57 @@ $$""".format(name, ' '.join(options)), name, password, password)
         90600
         """
         return Postgresql.postgres_version_to_int(pg_version + '.0')
+
+    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()
+                for opt in data.split('" "'):
+                    if '=' in opt and opt.startswith('--'):
+                        name, val = opt.split('=', 1)
+                        result[name.strip('-')] = val.rstrip('"\n')
+        except IOError:
+            logger.exception('Error when reading postmaster.opts')
+        return result
+
+    def single_user_mode(self, command=None, options=None):
+        """run a given command in a single-user mode. If the command is empty - then just start and stop"""
+        cmd = [self._pgcommand('postgres'), '--single', '-D', self._data_dir]
+        for opt, val in sorted((options or {}).items()):
+            cmd.extend(['-c', '{0}={1}'.format(opt, val)])
+        # need a database name to connect
+        cmd.append(self._database)
+        p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
+        if p:
+            if command:
+                p.communicate('{0}\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_' + self.wal_name, 'archive_status')
+        try:
+            for f in os.listdir(status_dir):
+                path = os.path.join(status_dir, f)
+                try:
+                    if os.path.islink(path):
+                        os.unlink(path)
+                    elif os.path.isfile(path):
+                        os.remove(path)
+                except OSError:
+                    logger.exception('Unable to remove %s', path)
+        except OSError:
+            logger.exception('Unable to list %s', status_dir)
+
+    def fix_cluster_state(self):
+        self.cleanup_archive_status()
+
+        # Start in a single user mode and stop to produce a clean shutdown
+        opts = self.read_postmaster_opts()
+        opts.update({'archive_mode': 'on', 'archive_command': 'false'})
+        if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf):
+            os.unlink(self._recovery_conf)
+        return self.single_user_mode(options=opts) == 0 or None

tests/test_ha.py

@@ -193,6 +193,15 @@ class TestHa(unittest.TestCase):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEquals(self.ha.run_cycle(), 'running pg_rewind from leader')
 
+    @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True))
+    @patch.object(Postgresql, 'fix_cluster_state', Mock())
+    def test_single_user_after_recover_failed(self):
+        self.p.controldata = lambda: {'Database cluster state': 'in production'}
+        self.p.is_running = false
+        self.p.follow = false
+        self.assertEquals(self.ha.run_cycle(), 'starting as a secondary')
+        self.assertEquals(self.ha.run_cycle(), 'fixing cluster state in a single user mode')
+
     @patch('sys.exit', return_value=1)
     @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True))
     def test_sysid_no_match(self, exit_mock):

tests/test_postgresql.py

@@ -327,10 +327,10 @@ class TestPostgresql(unittest.TestCase):
     @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True))
     def test__get_local_timeline_lsn(self):
         self.p.trigger_check_diverged_lsn()
-        with patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down'})):
-            self.p.rewind_needed_and_possible(self.leader)
         with patch.object(Postgresql, 'controldata',
-                          Mock(return_value={'Database cluster state': 'shut down in recovery'})):
+                          Mock(return_value={'Database cluster state': 'shut down in recovery',
+                                             'Minimum recovery ending location': '0/0',
+                                             "Min recovery ending loc's timeline": '0'})):
             self.p.rewind_needed_and_possible(self.leader)
         with patch.object(Postgresql, 'is_running', Mock(return_value=True)):
             with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(False, ), Exception])):
@@ -883,3 +883,40 @@ class TestPostgresql(unittest.TestCase):
     def test_terminate_starting_postmaster(self):
         self.p.terminate_starting_postmaster(123)
         self.p.terminate_starting_postmaster(123)
+
+    def test_read_postmaster_opts(self):
+        m = mock_open(read_data='/usr/lib/postgres/9.6/bin/postgres "-D" "data/postgresql0" \
+"--listen_addresses=127.0.0.1" "--port=5432" "--hot_standby=on" "--wal_level=hot_standby" \
+"--wal_log_hints=on" "--max_wal_senders=5" "--max_replication_slots=5"\n')
+        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
+            data = self.p.read_postmaster_opts()
+            self.assertEqual(data, dict())
+
+    @patch('subprocess.Popen')
+    @patch.object(builtins, 'open', Mock(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(command="CHECKPOINT"), 0)
+        subprocess_popen_mock.return_value = None
+        self.assertEquals(self.p.single_user_mode(), 1)
+
+    @patch('os.listdir', Mock(side_effect=[OSError, ['a', 'b']]))
+    @patch('os.unlink', Mock(side_effect=OSError))
+    @patch('os.remove', Mock())
+    @patch('os.path.islink', Mock(side_effect=[True, False]))
+    @patch('os.path.isfile', Mock(return_value=True))
+    def test_cleanup_archive_status(self):
+        self.p.cleanup_archive_status()
+        self.p.cleanup_archive_status()
+
+    @patch('os.unlink', Mock())
+    @patch('os.path.isfile', Mock(return_value=True))
+    @patch.object(Postgresql, 'single_user_mode', Mock(return_value=0))
+    def test_fix_cluster_state(self):
+        self.assertTrue(self.p.fix_cluster_state())