ensure a stale Postgres does not become leader

helpers/etcd.py

@@ -79,13 +79,22 @@ class Etcd:
                 logger.info("Could not take out TTL lock: %s" % e)
             return False
 
-    def update_leader(self, value):
+    def update_leader(self, state_handler):
         try:
-            self.put_client_path("/leader", {"value": value, "ttl": self.ttl, "prevValue": value})
+            self.put_client_path("/leader", {"value": state_handler.name, "ttl": self.ttl, "prevValue": state_handler.name})
+            self.put_client_path("/optime/leader", {"value": state_handler.last_operation()})
         except urllib2.HTTPError:
-            logger.error("Error updating TTL on ETCD for primary.")
+            logger.error("Error updating leader lock and optime on ETCD for primary.")
             return False
 
+    def last_leader_operation(self):
+        try:
+            return int(self.get_client_path("/optime/leader")["node"]["value"])
+        except urllib2.HTTPError as e:
+            if e.code == 404:
+                logger.error("Error updating TTL on ETCD for primary.")
+                return None
+
     def leader_unlocked(self):
         try:
             self.get_client_path("/leader")

helpers/ha.py

@@ -22,7 +22,10 @@ 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 update_last_leader_operation(self):
+        return self.etcd.update_last_leader_operation(self.state_handler.last_operation)
 
     def is_unlocked(self):
         return self.etcd.leader_unlocked()
@@ -35,7 +38,7 @@ class Ha:
 
     def run_cycle(self):
         try:
-            if self.state_handler.is_healthy():
+            if self.state_handler.is_healthy(self.etcd.last_leader_operation()):
                 if self.is_unlocked():
                     if self.state_handler.is_healthiest_node(self.etcd.members()):
                         if self.acquire_lock():

helpers/postgresql.py

@@ -6,7 +6,6 @@ from urlparse import urlparse
 
 logger = logging.getLogger(__name__)
 
-
 class Postgresql:
 
     def __init__(self, config):
@@ -111,11 +110,21 @@ class Postgresql:
             options += " -c \"%s=%s\"" % (setting, value)
         return options
 
-    def is_healthy(self):
+    def is_healthy(self, last_leader_operation):
         if not self.is_running():
             logger.warning("Postgresql is not running.")
             return False
 
+        if self.is_leader():
+            return True
+
+        # 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
+
         return True
 
     def is_healthiest_node(self, members):
@@ -182,3 +191,6 @@ recovery_target_timeline = 'latest'
 
     def xlog_position(self):
         return self.query("SELECT pg_last_xlog_replay_location() - '0/0000000'::pg_lsn;").fetchone()[0]
+
+    def last_operation(self):
+        return self.query("SELECT pg_current_xlog_location() - '0/00000'::pg_lsn;").fetchone()[0]

postgres0.yml

@@ -7,6 +7,7 @@ postgresql:
   name: postgresql0
   listen: 127.0.0.1:5432
   data_dir: data/postgresql0
+  maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
   replication:
     username: replicator
     password: rep-pass

postgres1.yml

@@ -7,6 +7,7 @@ postgresql:
   name: postgresql1
   listen: 127.0.0.1:5433
   data_dir: data/postgresql1
+  maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
   replication:
     username: replicator
     password: rep-pass