Merge pull request #7 from zalando/features/refactoring

Merge changes from compose/governer

README.md

@@ -35,6 +35,29 @@ We provide a haproxy configuration, which will give your application a single en
 
 For a diagram of the high availability decision loop, see the included a PDF: [postgres-ha.pdf](https://github.com/compose/template-etcd-based-postgres-ha/blob/master/postgres-ha.pdf)
 
+## YAML Configuration
+
+For an example file, see `postgres0.yml`.  Below is an explanation of settings:
+
+* *loop_wait*: the number of seconds the loop will sleep
+
+* *etcd*
+  * *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd
+  * *ttl*: the TTL to acquire the leader lock.  Think of it as the length of time before automatic failover process is initiated.
+  * *host*: the host:port for the etcd endpoint
+
+* *postgresql*
+  * *name*: the name of the Postgres host, must be unique for the cluster
+  * *listen*: ip address + port that Postgres listening. Must be accessible from other nodes in the cluster if using streaming replication.
+  * *data_dir*: file path to initialize and store Postgres data files
+  * *maximum_lag_on_failover*: the maximum bytes a follower may lag before it is not eligible become leader
+  * *replication*
+    * *username*: replication username, user will be created during initialization
+    * *password*: replication password, user will be created during initialization
+    * *network*: network setting for replication in pg_hba.conf
+  * *recovery_conf*: configuration settings written to recovery.conf when configuring follower
+  * *parameters*: list of configuration settings for Postgres
+
 ## Requirements on a Mac
 
 Run the following on a Mac to install requirements:

governor.py

@@ -51,9 +51,12 @@ class Governor:
         self.postgresql = Postgresql(config['postgresql'], aws_host_address)
         self.ha = Ha(self.postgresql, self.etcd)
 
+    def touch_member(self):
+        return self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string)
+
     def initialize(self):
         # wait for etcd to be available
-        while not self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string):
+        while not self.touch_member():
             logging.info('waiting on etcd')
             time.sleep(5)
 
@@ -76,6 +79,7 @@ class Governor:
 
     def run(self):
         while True:
+            self.touch_member()
             logging.info(self.ha.run_cycle())
             time.sleep(self.nap_time)
 

haproxy_status.sh

@@ -7,10 +7,10 @@ do {
         if [ "${response}" == " f" ]
         then
                 echo "HTTP/1.1 200 OK"
-                echo "X-XLOG-POSITION: $(echo "SELECT pg_current_xlog_location();" | psql -t postgres 2> /dev/null | tr -d ' ' | head -n 1)"
+                echo "X-XLOG-POSITION: $(echo "SELECT pg_current_xlog_location();" | psql -t postgres --port $2 --host $1 2> /dev/null | tr -d ' ' | head -n 1)"
         else
                 echo "HTTP/1.1 503 Service unavailable"
-                echo "X-XLOG-POSITION: $(echo "SELECT pg_last_xlog_replay_location();" | psql -t postgres 2> /dev/null | tr -d ' ' | head -n 1)"
+                echo "X-XLOG-POSITION: $(echo "SELECT pg_last_xlog_replay_location();" | psql -t postgres --port $2 --host $1 2> /dev/null | tr -d ' ' | head -n 1)"
         fi
 
 } | nc -l $3; done

helpers/etcd.py

@@ -45,6 +45,8 @@ class Etcd:
                 time.sleep(3)
             elif ex:
                 raise ex
+            else:
+                break
 
         return response.json(), response.status_code
 
@@ -115,7 +117,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 +127,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=state_handler.name, 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

@@ -150,10 +150,12 @@ 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):
+        if (last_leader_operation or 0) - self.xlog_position() > self.config.get('maximum_lag_on_failover', 0):
+            return False
+
         for member in members:
             if member.hostname == self.name:
                 continue
@@ -266,3 +268,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]

postgres0.yml

@@ -9,6 +9,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

@@ -9,6 +9,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