Merge branch 'master' of github.com:compose/governor

Conflicts:
	governor.py
	helpers/etcd.py
	helpers/ha.py
	helpers/postgresql.py

helpers/etcd.py

@@ -115,7 +115,7 @@ class Etcd:
             raise CurrentLeaderError("Etcd is not responding properly")
 
     def touch_member(self, member, connection_string):
-        return self.put_client_path('/members/' + member, value=connection_string)
+        return self.put_client_path('/members/' + member, value=connection_string, ttl=self.ttl)
 
     def take_leader(self, value):
         return self.put_client_path('/leader', value=value, ttl=self.ttl)
@@ -125,12 +125,24 @@ class Etcd:
         ret or logger.info('Could not take out TTL lock')
         return ret
 
-    def update_leader(self, value):
-        return self.put_client_path('/leader', value=value, ttl=self.ttl, prevValue=value)
+    def update_leader(self, state_handler):
+        ret = self.put_client_path('/leader', value=value, ttl=self.ttl, prevValue=state_handler.name)
+        ret and self.put_client_path('/optime/leader', value=state_handler.last_operation())
+        return ret
 
     def race(self, path, value):
         return self.put_client_path(path, value=value, prevExist=False)
 
+    def last_leader_operation(self):
+        try:
+            response, status_code = self.get_client_path('/optime/leader')
+            if status_code == 404:
+                return None
+            return int(response['node']['value'])
+        except:
+            logger.exception('last_leader_operation')
+        raise EtcdError('Etcd is not responding properly')
+
     def delete_member(self, member):
         return self.delete_client_path('/members/' + member)
 

helpers/ha.py

@@ -21,7 +21,7 @@ class Ha:
         return self.etcd.attempt_to_acquire_leader(self.state_handler.name)
 
     def update_lock(self):
-        return self.etcd.update_leader(self.state_handler.name)
+        return self.etcd.update_leader(self.state_handler)
 
     def is_unlocked(self):
         return not (self.cluster.leader and self.cluster.leader.hostname)
@@ -50,7 +50,7 @@ class Ha:
                 self.load_cluster_from_etcd()
 
             if self.is_unlocked():
-                if self.state_handler.is_healthiest_node(self.cluster.members):
+                if self.state_handler.is_healthiest_node(self.etcd.last_leader_operation(), self.cluster.members):
                     if self.acquire_lock():
                         if not self.state_handler.is_leader():
                             self.state_handler.promote()

helpers/postgresql.py

@@ -141,10 +141,16 @@ class Postgresql:
         if not self.is_running():
             logger.warning('Postgresql is not running.')
             return False
-
         return True
 
-    def is_healthiest_node(self, members):
+    def is_healthiest_node(self, last_leader_operation, members):
+        # this should only happen on initialization
+        if last_leader_operation is None:
+            return True
+
+        if last_leader_operation - self.xlog_position() > self.config['maximum_lag_on_failover']:
+            return False
+
         for member in members:
             if member.hostname == self.name:
                 continue
@@ -242,3 +248,6 @@ primary_conninfo = '{}'
                            WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots
                            WHERE slot_name = %s)""", slot, slot)
         self.members = members
+
+    def last_operation(self):
+        return self.query("SELECT pg_current_xlog_location() - '0/00000'::pg_lsn").fetchone()[0]