Respond 200 to master health-check only if update_lock was successful (#713)

If Patroni gets partitioned it starts receiving stale information from DCS.
We can't use this information to determine that we have the leader key.
Instead, we will record in Ha object the actual state of acquire/update lock and report as a leader only if it was successful.

P.S. despite responding with 200 on `GET /master` postgres was still running read-only.

patroni/api.py

@@ -92,7 +92,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             return response.get('role') == 'replica' and not patroni.noloadbalance
 
         if cluster:  # dcs available
-            if cluster.leader and cluster.leader.name == patroni.postgresql.name:  # is_leader
+            if patroni.ha.is_leader():
                 status_code = 200 if 'master' in path else 503
             elif 'role' not in response:
                 status_code = 503

patroni/ha.py

@@ -57,6 +57,8 @@ class Ha(object):
         self.dcs = patroni.dcs
         self.cluster = None
         self.old_cluster = None
+        self._is_leader = False
+        self._is_leader_lock = RLock()
         self._was_paused = False
         self._leader_timeline = None
         self.recovering = False
@@ -87,6 +89,14 @@ class Ha(object):
     def is_paused(self):
         return self.check_mode('pause')
 
+    def is_leader(self):
+        with self._is_leader_lock:
+            return self._is_leader
+
+    def set_is_leader(self, value):
+        with self._is_leader_lock:
+            self._is_leader = value
+
     def load_cluster_from_dcs(self):
         cluster = self.dcs.get_cluster()
 
@@ -98,7 +108,9 @@ class Ha(object):
         self._leader_timeline = None if cluster.is_unlocked() else cluster.leader.timeline
 
     def acquire_lock(self):
-        return self.dcs.attempt_to_acquire_leader()
+        ret = self.dcs.attempt_to_acquire_leader()
+        self.set_is_leader(ret)
+        return ret
 
     def update_lock(self, write_leader_optime=False):
         last_operation = None
@@ -108,6 +120,7 @@ class Ha(object):
             except Exception:
                 logger.exception('Exception when called state_handler.last_operation()')
         ret = self.dcs.update_leader(last_operation)
+        self.set_is_leader(ret)
         if ret:
             self.watchdog.keepalive()
         return ret
@@ -981,6 +994,7 @@ class Ha(object):
             logger.error('Cancelling bootstrap because watchdog activation failed')
             self.cancel_initialization()
         self.dcs.take_leader()
+        self.set_is_leader(True)
         self.state_handler.call_nowait(ACTION_ON_START)
         self.load_cluster_from_dcs()
         return 'initialized a new cluster'

tests/test_api.py

@@ -51,6 +51,10 @@ class MockHa(object):
     state_handler = MockPostgresql()
     watchdog = MockWatchdog()
 
+    @staticmethod
+    def is_leader():
+        return False
+
     @staticmethod
     def reinitialize(_):
         return 'reinitialize'
@@ -152,7 +156,7 @@ class TestRestApiHandler(unittest.TestCase):
             MockRestApiServer(RestApiHandler, 'GET /synchronous')
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
             MockRestApiServer(RestApiHandler, 'GET /asynchronous')
-        MockPatroni.dcs.cluster.leader.name = MockPostgresql.name
+        MockPatroni.ha.is_leader = Mock(return_value=True)
         MockRestApiServer(RestApiHandler, 'GET /replica')
         MockPatroni.dcs.cluster = None
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):

tests/test_patroni.py

@@ -103,6 +103,7 @@ class TestPatroni(unittest.TestCase):
 
     @patch('patroni.config.Config.save_cache', Mock())
     @patch('patroni.config.Config.reload_local_configuration', Mock(return_value=True))
+    @patch('patroni.ha.Ha.is_leader', Mock(return_value=True))
     @patch.object(Postgresql, 'state', PropertyMock(return_value='running'))
     @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
     def test_run(self):