Bugfix/fast recover (#300)

* reap children before and after running HA loop

When the Patroni is running in a docker container with the pid=1 it is
also responsible for reaping of all dead processes. We can't call
os.waitpid immediately after receiving SIGCHLD because it breaks
subprocess module. It simply stops receiving exit codes of the processes
it executes because these processes. That's why we just registering the
fact of receiving SIGCHLD and reaping children only after execution of
HA loop.
If the postmaster was dying for some reason, Patroni was able to detect
this fact only on the next iteration of HA loop, because zombie
processes where still there and it was possible to send 0 signal to it.
To avoid such situation we should also reap all dead processes before
executing HA loop.

* Don't rely on _cursor_holder when closing connection

it could happen that connection has been opened but not cursor...

* Don't "retry" when fetching current xlog location and it fails

On every iteration of HA loop we are updaing member key in DCS and among
other data there is current xlog location stored in the value.
If the postgres has died for some reason it is not possible to fetch
xlog position and we are just wasting retry_timeout/2 = 5 seconds there.
If this information will be missing from DCS during period of one HA
loop nothing should break. Patroni is not relying on this information
anyway. When it is doing manual or automatic failover it aways
communicates with other nodes directly to get the most fresh
infomation.

* Don't try to update leader optime when postgres is not 100% healthy

`update_lock` method is not only doing update of the leader lock but
also writes the most recent value of xlog position into optime/leader
key. If you know that postgres can be not 100% healthy because it is in
process of restart or recover we should not try to fetch current xlog
position and update 'optime/leader'. Previously we were using
`AsyncExecutor.busy` property for avoiding of such action, but I think
we should be more excpilicit and do the update only if we know that
postgres is 100% healty.

patroni/__init__.py

@@ -103,6 +103,8 @@ class Patroni(object):
                 if self.config.reload_local_configuration():
                     self.reload_config()
 
+            reap_children()
+
             logger.info(self.ha.run_cycle())
 
             cluster = self.dcs.cluster

patroni/ha.py

@@ -40,9 +40,9 @@ class Ha(object):
     def acquire_lock(self):
         return self.dcs.attempt_to_acquire_leader()
 
-    def update_lock(self):
+    def update_lock(self, write_leader_optime=False):
         ret = self.dcs.update_leader()
-        if ret and not self._async_executor.busy:
+        if ret and write_leader_optime:
             try:
                 self.dcs.write_leader_optime(self.state_handler.last_operation())
             except:
@@ -67,7 +67,7 @@ class Ha(object):
             data['pending_restart'] = True
         if not self._async_executor.busy and data['state'] in ['running', 'restarting', 'starting']:
             try:
-                data['xlog_location'] = self.state_handler.xlog_position()
+                data['xlog_location'] = self.state_handler.xlog_position(retry=False)
             except:
                 pass
         if self.patroni.scheduled_restart:
@@ -428,7 +428,7 @@ class Ha(object):
                 self.dcs.reset_cluster()
                 return 'removed leader lock because postgres is not running as master'
 
-            if self.update_lock():
+            if self.update_lock(True):
                 return self.enforce_master_role('no action.  i am the leader with the lock',
                                                 'promoted self to leader because i had the session lock')
             else:

patroni/postgresql.py

@@ -318,9 +318,10 @@ class Postgresql(object):
         return self._cursor_holder
 
     def close_connection(self):
-        if self._cursor_holder and self._cursor_holder.connection and self._cursor_holder.connection.closed == 0:
-            self._cursor_holder.connection.close()
+        if self._connection and self._connection.closed == 0:
+            self._connection.close()
             logger.info("closed patroni connection to the postgresql cluster")
+        self._cursor_holder = self._connection = None
 
     def _query(self, sql, *params):
         cursor = None
@@ -888,11 +889,12 @@ BEGIN
 END;
 $$""".format(name, ' '.join(options)), name, password, password)
 
-    def xlog_position(self):
-        return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
-                                                               THEN pg_last_xlog_replay_location()
-                                                               ELSE pg_current_xlog_location()
-                                                          END, '0/0')::bigint""").fetchone()[0]
+    def xlog_position(self, retry=True):
+        stmt = """SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
+                                                    THEN pg_last_xlog_replay_location()
+                                                    ELSE pg_current_xlog_location()
+                                               END, '0/0')::bigint"""
+        return (self.query(stmt) if retry else self._query(stmt)).fetchone()[0]
 
     def load_replication_slots(self):
         if self.use_slots and self._schedule_load_slots:

tests/test_ha.py

@@ -139,7 +139,7 @@ class TestHa(unittest.TestCase):
 
     def test_update_lock(self):
         self.p.last_operation = Mock(side_effect=PostgresException(''))
-        self.assertTrue(self.ha.update_lock())
+        self.assertTrue(self.ha.update_lock(True))
 
     def test_touch_member(self):
         self.p.xlog_position = Mock(side_effect=Exception)