Merge pull request #4 from zalando/feature/lost-dcs-data

Run is_healthiest_node against last know healthy configuration of cluster

helpers/ha.py

@@ -12,9 +12,17 @@ class Ha:
         self.state_handler = state_handler
         self.dcs = etcd
         self.cluster = None
+        self.old_cluster = None
 
     def load_cluster_from_dcs(self):
-        self.cluster = self.dcs.get_cluster()
+        cluster = self.dcs.get_cluster()
+
+        # We want to keep the state of cluster when it was healhy
+        if cluster.is_unlocked() and self.cluster and not self.cluster.is_unlocked():
+            self.old_cluster = self.cluster
+        if not self.old_cluster:
+            self.old_cluster = cluster
+        self.cluster = cluster
 
     def acquire_lock(self):
         return self.dcs.attempt_to_acquire_leader()
@@ -46,7 +54,7 @@ class Ha:
                 self.load_cluster_from_dcs()
 
             if self.cluster.is_unlocked():
-                if self.state_handler.is_healthiest_node(self.cluster):
+                if self.state_handler.is_healthiest_node(self.old_cluster):
                     if self.acquire_lock():
                         if self.state_handler.is_leader() or self.state_handler.is_promoted:
                             return 'acquired session lock as a leader'

helpers/postgresql.py

@@ -345,7 +345,10 @@ class Postgresql:
                 member_cursor.close()
                 member_conn.close()
                 logger.error([self.name, member.name, row])
-                if not row[0] or row[1] < 0:
+                if not row[0]:
+                    logger.warning('Master (%s) is still alive', member.name)
+                    return False
+                if row[1] < 0:
                     return False
             except psycopg2.Error:
                 continue

tests/test_ha.py

@@ -60,6 +60,10 @@ def dead_etcd():
     raise DCSError('Etcd is not responding properly')
 
 
+def get_unlocked_cluster():
+    return Cluster(False, None, None, [])
+
+
 class TestHa(unittest.TestCase):
 
     def __init__(self, method_name='runTest'):
@@ -74,9 +78,15 @@ class TestHa(unittest.TestCase):
         self.e = Etcd('foo', {'ttl': 30, 'host': 'remotehost:2379', 'scope': 'test'})
         self.ha = Ha(self.p, self.e)
         self.ha.load_cluster_from_dcs()
-        self.ha.cluster = Cluster(False, None, None, [])
+        self.ha.cluster = get_unlocked_cluster()
         self.ha.load_cluster_from_dcs = nop
 
+    def test_load_cluster_from_dcs(self):
+        ha = Ha(self.p, self.e)
+        ha.load_cluster_from_dcs()
+        self.e.get_cluster = get_unlocked_cluster
+        ha.load_cluster_from_dcs()
+
     def test_start_as_slave(self):
         self.p.is_healthy = false
         self.assertEquals(self.ha.run_cycle(), 'started as a secondary')

tests/test_postgresql.py

@@ -23,10 +23,6 @@ def false(*args, **kwargs):
     return False
 
 
-def xlog_position():
-    return 1
-
-
 class MockCursor:
 
     def __init__(self):
@@ -44,9 +40,12 @@ class MockCursor:
         elif sql.startswith('SELECT pg_current_xlog_location()'):
             self.results = [(0,)]
         elif sql.startswith('SELECT pg_is_in_recovery(), %s'):
-            if params[0][0] != 0:
+            if params[0][0] == 1:
                 raise psycopg2.OperationalError()
-            self.results = [(False, 0)]
+            elif params[0][0] == 2:
+                self.results = [(True, -1)]
+            else:
+                self.results = [(False, 0)]
         elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'):
             self.results = [(0,)]
         elif sql.startswith('SELECT pg_is_in_recovery()'):
@@ -187,8 +186,10 @@ class TestPostgresql(unittest.TestCase):
         self.assertTrue(self.p.is_healthiest_node(cluster))
         self.p.is_leader = false
         self.assertFalse(self.p.is_healthiest_node(cluster))
-        self.p.xlog_position = xlog_position
+        self.p.xlog_position = lambda: 1
         self.assertTrue(self.p.is_healthiest_node(cluster))
+        self.p.xlog_position = lambda: 2
+        self.assertFalse(self.p.is_healthiest_node(cluster))
         self.p.config['maximum_lag_on_failover'] = -2
         self.assertFalse(self.p.is_healthiest_node(cluster))