From f53c369c69fa11a98a9765138f537479b22310ec Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sat, 23 May 2015 20:21:13 +0200
Subject: [PATCH 01/12] Connect timeout 5 seconds for psycopg2

Refactored query and is_healthiest_node methods
query method wont retry more then 3 times if OperationalError raised
is_healthiest_node takes into account that one of the nodes could be
still a master
---
 helpers/postgresql.py    | 95 ++++++++++++++++++++++------------------
 tests/test_postgresql.py | 35 ++++++++++-----
 2 files changed, 78 insertions(+), 52 deletions(-)

diff --git a/helpers/postgresql.py b/helpers/postgresql.py
index 8d32d65d..11be38c0 100644
--- a/helpers/postgresql.py
+++ b/helpers/postgresql.py
@@ -18,63 +18,69 @@ logger = logging.getLogger(__name__)
 def parseurl(url):
     r = urlparse(url)
     return {
-        'hostname': r.hostname,
+        'host': r.hostname,
         'port': r.port or 5432,
-        'username': r.username,
+        'user': r.username,
         'password': r.password,
+        'database': r.path[1:],
+        'fallback_application_name': 'Governor',
+        'connect_timeout': 5,
     }
 
 
 class Postgresql:
 
     def __init__(self, config):
+        self.config = config
         self.name = config['name']
         self.listen_addresses, self.port = config['listen'].split(':')
         self.data_dir = config['data_dir']
         self.replication = config['replication']
         self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf')
+        self.pid_path = os.path.join(self.data_dir, 'postmaster.pid')
         self._pg_ctl = 'pg_ctl -w -D ' + self.data_dir
 
-        self.config = config
-
-        connect_address = config.get('connect_address', config['listen']) or config['listen']
+        self.local_address = self.get_local_address()
+        connect_address = config.get('connect_address', None) or self.local_address
         self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format(
             connect_address=connect_address, **self.replication)
 
-        self.conn = None
-        self.cursor_holder = None
+        self._connection = None
+        self._cursor_holder = None
         self.members = []  # list of already existing replication slots
 
-    def cursor(self):
-        if not self.cursor_holder:
-            self.conn = psycopg2.connect('postgres://{}:{}/postgres'.format(
-                self.listen_addresses.split(',')[0].strip(), self.port))
-            self.conn.autocommit = True
-            self.cursor_holder = self.conn.cursor()
+    def get_local_address(self):
+        # TODO: try to get unix_socket_directory from postmaster.pid
+        return self.listen_addresses.split(',')[0].strip() + ':' + self.port
 
-        return self.cursor_holder
+    def connection(self):
+        if not self._connection or self._connection.closed:
+            self._connection = psycopg2.connect('postgres://{}/postgres'.format(self.local_address))
+            self._connection.autocommit = True
+        return self._connection
+
+    def _cursor(self):
+        if not self._cursor_holder or self._cursor_holder.closed != 0:
+            self._cursor_holder = self.connection().cursor()
+        return self._cursor_holder
 
     def disconnect(self):
-        try:
-            self.conn.close()
-        except:
-            logger.exception('Error disconnecting')
+        self._connection and self._connection.close()
+        self._connection = self._cursor_holder = None
 
     def query(self, sql, *params):
         max_attempts = 0
         while True:
             try:
-                self.cursor().execute(sql, params)
-                break
-            except psycopg2.OperationalError as e:
-                if self.conn:
-                    self.disconnect()
-                self.cursor_holder = None
-                if max_attempts > 4:
-                    raise e
+                cursor = self._cursor()
+                cursor.execute(sql, params)
+                return cursor
+            except (psycopg2.OperationalError, psycopg2.InterfaceError) as e:
+                self.disconnect()
                 max_attempts += 1
+                if max_attempts >= 3:
+                    raise e
                 time.sleep(5)
-        return self.cursor()
 
     def data_directory_empty(self):
         return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []
@@ -90,11 +96,11 @@ class Postgresql:
         pgpass = 'pgpass'
         with open(pgpass, 'w') as f:
             os.fchmod(f.fileno(), 0o600)
-            f.write('{hostname}:{port}:*:{username}:{password}\n'.format(**r))
+            f.write('{host}:{port}:*:{user}:{password}\n'.format(**r))
 
         try:
             os.environ['PGPASSFILE'] = pgpass
-            return os.system('pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format(
+            return os.system('pg_basebackup -R -D {data_dir} --host={host} --port={port} -U {user}'.format(
                 data_dir=self.data_dir, **r)) == 0
         finally:
             os.environ.pop('PGPASSFILE')
@@ -111,10 +117,9 @@ class Postgresql:
             logger.error('Cannot start PostgreSQL because one is already running.')
             return False
 
-        pid_path = os.path.join(self.data_dir, 'postmaster.pid')
-        if os.path.exists(pid_path):
-            os.remove(pid_path)
-            logger.info('Removed %s', pid_path)
+        if os.path.exists(self.pid_path):
+            os.remove(self.pid_path)
+            logger.info('Removed %s', self.pid_path)
 
         ret = os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0
         ret and self.load_replication_slots()
@@ -142,6 +147,9 @@ class Postgresql:
         return True
 
     def is_healthiest_node(self, cluster):
+        if self.is_leader():
+            return True
+
         if cluster.last_leader_operation - self.xlog_position() > self.config.get('maximum_lag_on_failover', 0):
             return False
 
@@ -149,18 +157,19 @@ class Postgresql:
             if member.hostname == self.name:
                 continue
             try:
-                member_conn = psycopg2.connect(member.address)
+                member_conn = psycopg2.connect(parseurl(member.address))
                 member_conn.autocommit = True
                 member_cursor = member_conn.cursor()
                 member_cursor.execute(
-                    "SELECT %s - (pg_last_xlog_replay_location() - '0/0000000'::pg_lsn)", (self.xlog_position(), ))
-                xlog_diff = member_cursor.fetchone()[0]
-                logger.info([self.name, member.hostname, xlog_diff])
+                    "SELECT pg_is_in_recovery(), %s - (pg_last_xlog_replay_location() - '0/0000000'::pg_lsn)",
+                    (self.xlog_position(), ))
+                row = member_cursor.fetchone()
                 member_cursor.close()
                 member_conn.close()
-                if xlog_diff < 0:
+                logger.error([self.name, member.hostname, row])
+                if not row[0] or row[1] < 0:
                     return False
-            except psycopg2.OperationalError:
+            except psycopg2.Error:
                 continue
         return True
 
@@ -171,7 +180,7 @@ class Postgresql:
     @staticmethod
     def primary_conninfo(leader_url):
         r = parseurl(leader_url)
-        return 'user={username} password={password} host={hostname} port={port} sslmode=prefer sslcompression=1'.format(**r)
+        return 'user={user} password={password} host={host} port={port} sslmode=prefer sslcompression=1'.format(**r)
 
     def check_recovery_conf(self, leader):
         if not os.path.isfile(self.recovery_conf):
@@ -218,7 +227,9 @@ primary_conninfo = '{}'
             self.replication['username']), self.replication['password'])
 
     def xlog_position(self):
-        return self.query("SELECT pg_last_xlog_replay_location() - '0/0000000'::pg_lsn").fetchone()[0]
+        return self.query("""SELECT CASE WHEN pg_is_in_recovery()
+                                         THEN pg_last_xlog_replay_location() - '0/0000000'::pg_lsn
+                                         ELSE pg_current_xlog_location() - '0/00000'::pg_lsn END""").fetchone()[0]
 
     def load_replication_slots(self):
         cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")
@@ -239,4 +250,4 @@ primary_conninfo = '{}'
         self.members = members
 
     def last_operation(self):
-        return self.query("SELECT pg_current_xlog_location() - '0/00000'::pg_lsn").fetchone()[0]
+        return self.xlog_position()
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 297bd929..5442e58c 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -11,9 +11,18 @@ def os_system(cmd):
     return 0
 
 
+def false(*args, **kwargs):
+    return False
+
+
+def xlog_position():
+    return 1
+
+
 class MockCursor:
 
     def __init__(self):
+        self.closed = 0
         self.current = 0
         self.results = []
 
@@ -21,15 +30,19 @@ class MockCursor:
         if sql.startswith('blabla'):
             raise psycopg2.OperationalError()
         elif sql.startswith('SELECT slot_name'):
-            self.results = [('blabla'), ('foobar')]
+            self.results = [('blabla',), ('foobar',)]
         elif sql.startswith('SELECT pg_current_xlog_location()'):
             self.results = [(0,)]
-        elif sql.startswith('SELECT %s - (pg_last_xlog_replay_location()'):
-            self.results = [(0,)]
-        elif sql.startswith('SELECT pg_last_xlog_replay_location()'):
+        elif sql.startswith('SELECT pg_is_in_recovery(), %s'):
+            if params[0][0] != 0:
+                raise psycopg2.OperationalError()
+            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()'):
             self.results = [(False, )]
+        elif sql.startswith('SELECT to_char(pg_postmaster_start_time'):
+            self.results = [('', True, '', '', '', False)]
         else:
             self.results = []
 
@@ -48,13 +61,13 @@ class MockConnect:
 
     def __init__(self):
         self.autocommit = False
+        self.closed = False
 
     def cursor(self):
         return MockCursor()
 
     def close(self):
-        if not self.autocommit:
-            raise psycopg2.OperationalError()
+        pass
 
 
 def psycopg2_connect(*args, **kwargs):
@@ -115,17 +128,19 @@ class TestPostgresql(unittest.TestCase):
 
     def test_query(self):
         self.p.query('select 1')
-        self.p.conn.autocommit = False
         self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
-        self.p.query('select %s', 1)
 
     def test_is_healthiest_node(self):
         leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', 28)
         me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', 28)
         other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', 28)
-        cluster = Cluster(leader, 0, [leader, me, other])
+        cluster = Cluster(leader, 0, [me, other, leader])
         self.assertTrue(self.p.is_healthiest_node(cluster))
-        self.p.config['maximum_lag_on_failover'] = -1
+        self.p.is_leader = false
+        self.assertFalse(self.p.is_healthiest_node(cluster))
+        self.p.xlog_position = xlog_position
+        self.assertTrue(self.p.is_healthiest_node(cluster))
+        self.p.config['maximum_lag_on_failover'] = -2
         self.assertFalse(self.p.is_healthiest_node(cluster))
 
     def test_is_leader(self):

From d202f72a429bd67008a6ca6e63859b09fda00bcd Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sat, 23 May 2015 20:29:51 +0200
Subject: [PATCH 02/12] Added simple web server which shows current state of
 postgres

Server always returns json which contains some status of postgres:
* instance type: master/slave
* xlog location for master
* xlog received_location, replayed_location
This server could be used by haproxy
GET / returns 200 if there is postgres master behind governor.
GET /slave returns 200 is there is postgres slave behind governor
In all other cases it returns 503
Currently server always listening on 0.0.0.0:8080, but it should be
configurable. In the next versions this rest api could also report some
status of etcd and could be used by is_healthiest_node method
---
 governor.py       |  2 ++
 helpers/api.py    | 69 +++++++++++++++++++++++++++++++++++++++++++++++
 tests/test_api.py | 66 +++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 137 insertions(+)
 create mode 100644 helpers/api.py
 create mode 100644 tests/test_api.py

diff --git a/governor.py b/governor.py
index d12c7093..27102d88 100755
--- a/governor.py
+++ b/governor.py
@@ -7,6 +7,7 @@ import sys
 import time
 import yaml
 
+from helpers.api import RestApiServer
 from helpers.etcd import Etcd
 from helpers.postgresql import Postgresql
 from helpers.ha import Ha
@@ -81,6 +82,7 @@ def main():
     governor = Governor(config)
     try:
         governor.initialize()
+        RestApiServer(governor).start()
         governor.run()
     finally:
         governor.postgresql.stop()
diff --git a/helpers/api.py b/helpers/api.py
new file mode 100644
index 00000000..6269a83c
--- /dev/null
+++ b/helpers/api.py
@@ -0,0 +1,69 @@
+import json
+import logging
+import psycopg2
+import sys
+
+from threading import Thread
+
+if sys.hexversion >= 0x03000000:
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+else:
+    from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
+
+
+logger = logging.getLogger(__name__)
+
+
+class RestApiHandler(BaseHTTPRequestHandler):
+
+    def do_GET(self):
+        try:
+            response = self.get_postgresql_status()
+        except (psycopg2.OperationalError, psycopg2.InterfaceError):
+            response = {'running': False}
+
+        path = '/master' if self.path == '/' else self.path
+        status_code = 200 if response['running'] and response['role'] in path else 503
+
+        self.send_response(status_code)
+        self.send_header('Content-Type', 'application/json')
+        self.end_headers()
+        self.wfile.write(json.dumps(response).encode('utf-8'))
+
+    def get_postgresql_status(self):
+        if not self.server.governor.postgresql.is_running():
+            return {'running': False}
+        cursor = self.server.cursor()
+        cursor.execute("""SELECT to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),
+                                 pg_is_in_recovery(),
+                                 pg_current_xlog_location(),
+                                 pg_last_xlog_receive_location(),
+                                 pg_last_xlog_replay_location(),
+                                 pg_is_in_recovery() AND pg_is_xlog_replay_paused()""")
+        row = cursor.fetchone()
+        return {
+            'running': True,
+            'postmaster_start_time': row[0],
+            'role': 'slave' if row[1] else 'master',
+            'xlog': ({
+                'received_location': row[3],
+                'replayed_location': row[4],
+                'paused': row[5]} if row[1] else {
+                'location': row[2]
+            })
+        }
+
+
+class RestApiServer(HTTPServer, Thread):
+
+    def __init__(self, governor, listen_address='0.0.0.0', listen_port=8080):
+        HTTPServer.__init__(self, (listen_address, listen_port), RestApiHandler)
+        Thread.__init__(self, target=self.serve_forever)
+        self.governor = governor
+        self._cursor_holder = None
+        self.daemon = True
+
+    def cursor(self):
+        if not self._cursor_holder or self._cursor_holder.closed != 0:
+            self._cursor_holder = self.governor.postgresql.connection().cursor()
+        return self._cursor_holder
diff --git a/tests/test_api.py b/tests/test_api.py
new file mode 100644
index 00000000..6b117b48
--- /dev/null
+++ b/tests/test_api.py
@@ -0,0 +1,66 @@
+import psycopg2
+import sys
+import unittest
+
+from helpers.api import RestApiHandler, RestApiServer
+from test_postgresql import psycopg2_connect
+
+if sys.hexversion >= 0x03000000:
+    from io import BytesIO as IO
+else:
+    from StringIO import StringIO as IO
+
+
+def false(*args, **kwargs):
+    return False
+
+
+def throws(*args, **kwargs):
+    raise psycopg2.OperationalError()
+
+
+class MockPostgresql:
+
+    def connection(self):
+        return psycopg2_connect()
+
+    def is_running(self):
+        return True
+
+
+class MockGovernor:
+
+    def __init__(self):
+        self.postgresql = MockPostgresql()
+
+
+class MockRequest:
+
+    def __init__(self, path):
+        self.path = path
+
+    def makefile(self, *args, **kwargs):
+        return IO(self.path)
+
+
+class MockRestApiServer(RestApiServer):
+
+    def __init__(self, Handler, path, *args):
+        self.governor = MockGovernor()
+        if len(args) > 0:
+            self.governor.postgresql.is_running = args[0]
+        self._cursor_holder = None
+        Handler(MockRequest(path), ('0.0.0.0', 8080), self)
+
+
+class TestRestApiHandler(unittest.TestCase):
+
+    def __init__(self, method_name='runTest'):
+        super(TestRestApiHandler, self).__init__(method_name)
+
+    def test_do_GET(self):
+        MockRestApiServer(RestApiHandler, b'GET /')
+        MockRestApiServer(RestApiHandler, b'GET /', throws)
+
+    def test_get_postgresql_status(self):
+        MockRestApiServer(RestApiHandler, b'GET /', false)

From ce267d9f1e07c531ffa6fb6844e9a1ce9e2b465f Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sun, 24 May 2015 09:01:23 +0200
Subject: [PATCH 03/12] query method will retry only in case of communication
 error

---
 helpers/ha.py            |  4 ++--
 helpers/postgresql.py    | 15 +++++++++++----
 tests/test_postgresql.py | 12 ++++++++++--
 3 files changed, 23 insertions(+), 8 deletions(-)

diff --git a/helpers/ha.py b/helpers/ha.py
index 74c6a892..773c9b57 100644
--- a/helpers/ha.py
+++ b/helpers/ha.py
@@ -1,7 +1,7 @@
 import logging
 
 from helpers.errors import EtcdError, HealthiestMemberError
-from psycopg2 import OperationalError
+from psycopg2 import InterfaceError, OperationalError
 
 logger = logging.getLogger(__name__)
 
@@ -92,7 +92,7 @@ class Ha:
             if self.state_handler.is_leader():
                 self.state_handler.demote(None)
                 return 'demoted self because etcd is not accessible and i was a leader'
-        except OperationalError:
+        except (InterfaceError, OperationalError):
             logger.error('Error communicating with Postgresql.  Will try again')
         except HealthiestMemberError:
             logger.error('failed to determine healthiest member fromt etcd')
diff --git a/helpers/postgresql.py b/helpers/postgresql.py
index 11be38c0..720d0775 100644
--- a/helpers/postgresql.py
+++ b/helpers/postgresql.py
@@ -54,13 +54,13 @@ class Postgresql:
         return self.listen_addresses.split(',')[0].strip() + ':' + self.port
 
     def connection(self):
-        if not self._connection or self._connection.closed:
+        if not self._connection or self._connection.closed != 0:
             self._connection = psycopg2.connect('postgres://{}/postgres'.format(self.local_address))
             self._connection.autocommit = True
         return self._connection
 
     def _cursor(self):
-        if not self._cursor_holder or self._cursor_holder.closed != 0:
+        if not self._cursor_holder or self._cursor_holder.closed:
             self._cursor_holder = self.connection().cursor()
         return self._cursor_holder
 
@@ -71,15 +71,22 @@ class Postgresql:
     def query(self, sql, *params):
         max_attempts = 0
         while True:
+            ex = None
             try:
                 cursor = self._cursor()
                 cursor.execute(sql, params)
                 return cursor
-            except (psycopg2.OperationalError, psycopg2.InterfaceError) as e:
+            except psycopg2.InterfaceError as e:
+                ex = e
+            except psycopg2.OperationalError as e:
+                if self._connection and self._connection.closed == 0:
+                    raise e
+                ex = e
+            if ex:
                 self.disconnect()
                 max_attempts += 1
                 if max_attempts >= 3:
-                    raise e
+                    raise ex
                 time.sleep(5)
 
     def data_directory_empty(self):
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 5442e58c..e0a21645 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -22,13 +22,15 @@ def xlog_position():
 class MockCursor:
 
     def __init__(self):
-        self.closed = 0
+        self.closed = False
         self.current = 0
         self.results = []
 
     def execute(self, sql, *params):
         if sql.startswith('blabla'):
             raise psycopg2.OperationalError()
+        elif sql.startswith('InterfaceError'):
+            raise psycopg2.InterfaceError()
         elif sql.startswith('SELECT slot_name'):
             self.results = [('blabla',), ('foobar',)]
         elif sql.startswith('SELECT pg_current_xlog_location()'):
@@ -61,7 +63,7 @@ class MockConnect:
 
     def __init__(self):
         self.autocommit = False
-        self.closed = False
+        self.closed = 0
 
     def cursor(self):
         return MockCursor()
@@ -128,6 +130,12 @@ class TestPostgresql(unittest.TestCase):
 
     def test_query(self):
         self.p.query('select 1')
+        self.assertRaises(psycopg2.InterfaceError, self.p.query, 'InterfaceError')
+        self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
+        self.p._connection.closed = 2
+        self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
+        self.p._connection.closed = 2
+        self.p.disconnect = false
         self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
 
     def test_is_healthiest_node(self):

From 8dd99170bfae0cb52ce349c7dab3eecc3ff590e3 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sun, 24 May 2015 15:40:36 +0200
Subject: [PATCH 04/12] Refactor exceptions handling in etcd.py and ha.py

update_leader can throw EtcdError exception
unused HealthiestMemberError exception is removed
---
 helpers/errors.py  |  4 ----
 helpers/etcd.py    | 52 ++++++++++++++++++++++++++++++----------------
 helpers/ha.py      |  4 +---
 tests/test_etcd.py | 44 ++++++++++++++++++++++++++++++++++-----
 tests/test_ha.py   |  1 +
 5 files changed, 75 insertions(+), 30 deletions(-)

diff --git a/helpers/errors.py b/helpers/errors.py
index dd1bc698..b86bd0f0 100644
--- a/helpers/errors.py
+++ b/helpers/errors.py
@@ -9,7 +9,3 @@ class EtcdError(Exception):
 
 class CurrentLeaderError(EtcdError):
     pass
-
-
-class HealthiestMemberError(EtcdError):
-    pass
diff --git a/helpers/etcd.py b/helpers/etcd.py
index e808f8d4..966f881b 100644
--- a/helpers/etcd.py
+++ b/helpers/etcd.py
@@ -2,12 +2,12 @@ import logging
 import requests
 import time
 
+from requests.exceptions import RequestException
 from collections import namedtuple
 from helpers.errors import CurrentLeaderError, EtcdError
 
 logger = logging.getLogger(__name__)
 
-
 Member = namedtuple('Member', 'hostname,address,ttl')
 
 
@@ -35,7 +35,7 @@ class Etcd:
                 response = requests.get(self.client_url(path))
                 if response.status_code == 200:
                     break
-            except Exception as e:
+            except RequestException as e:
                 logger.exception('get_client_path')
                 ex = e
 
@@ -54,15 +54,15 @@ class Etcd:
         try:
             response = requests.put(self.client_url(path), data=data)
             return response.status_code in [200, 201, 202, 204]
-        except:
+        except RequestException:
             logger.exception('PUT %s data=%s', path, data)
-            return False
+        raise EtcdError('Etcd is not responding properly')
 
     def delete_client_path(self, path):
         try:
             response = requests.delete(self.client_url(path))
             return response.status_code in [200, 202, 204]
-        except:
+        except RequestException:
             logger.exception('DELETE %s', path)
             return False
 
@@ -108,7 +108,7 @@ class Etcd:
                             leader = m
                             break
                     if not leader:
-                        leader = Member(leader['value'], None, None)
+                        leader = Member(node['value'], None, None)
 
                 return Cluster(leader, last_leader_operation, members)
             elif status_code == 404:
@@ -122,27 +122,43 @@ class Etcd:
         try:
             cluster = self.get_cluster()
             return None if cluster.is_unlocked() else cluster.leader
-        except:
-            raise CurrentLeaderError("Etcd is not responding properly")
+        except EtcdError:
+            raise CurrentLeaderError('Etcd is not responding properly')
 
-    def touch_member(self, member, connection_string):
-        return self.put_client_path('/members/' + member, value=connection_string, ttl=self.member_ttl)
+    def touch_member(self, member, connection_string, ttl=None):
+        try:
+            return self.put_client_path('/members/' + member, value=connection_string, ttl=ttl or self.member_ttl)
+        except EtcdError:
+            return False
 
     def take_leader(self, value):
-        return self.put_client_path('/leader', value=value, ttl=self.ttl)
+        try:
+            return self.put_client_path('/leader', value=value, ttl=self.ttl)
+        except EtcdError:
+            return False
 
     def attempt_to_acquire_leader(self, value):
-        ret = self.put_client_path('/leader', value=value, ttl=self.ttl, prevExist=False)
-        ret or logger.info('Could not take out TTL lock')
-        return ret
+        try:
+            ret = self.put_client_path('/leader', value=value, ttl=self.ttl, prevExist=False)
+            ret or logger.info('Could not take out TTL lock')
+            return ret
+        except EtcdError:
+            return False
 
     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
+        if self.put_client_path('/leader', value=state_handler.name, ttl=self.ttl, prevValue=state_handler.name):
+            try:
+                self.put_client_path('/optime/leader', value=state_handler.last_operation())
+            except EtcdError:
+                pass
+            return True
+        return False
 
     def race(self, path, value):
-        return self.put_client_path(path, value=value, prevExist=False)
+        try:
+            return self.put_client_path(path, value=value, prevExist=False)
+        except EtcdError:
+            return False
 
     def delete_member(self, member):
         return self.delete_client_path('/members/' + member)
diff --git a/helpers/ha.py b/helpers/ha.py
index 773c9b57..29979363 100644
--- a/helpers/ha.py
+++ b/helpers/ha.py
@@ -1,6 +1,6 @@
 import logging
 
-from helpers.errors import EtcdError, HealthiestMemberError
+from helpers.errors import EtcdError
 from psycopg2 import InterfaceError, OperationalError
 
 logger = logging.getLogger(__name__)
@@ -94,5 +94,3 @@ class Ha:
                 return 'demoted self because etcd is not accessible and i was a leader'
         except (InterfaceError, OperationalError):
             logger.error('Error communicating with Postgresql.  Will try again')
-        except HealthiestMemberError:
-            logger.error('failed to determine healthiest member fromt etcd')
diff --git a/tests/test_etcd.py b/tests/test_etcd.py
index c358e89e..8b3d60b2 100644
--- a/tests/test_etcd.py
+++ b/tests/test_etcd.py
@@ -17,28 +17,39 @@ class MockResponse:
         return json.loads(self.content)
 
 
+class MockPostgresql:
+    name = ''
+
+    def last_operation(self):
+        return 0
+
+
 def requests_get(url, **kwargs):
     if url.startswith('http://local'):
-        raise Exception()
+        raise requests.exceptions.RequestException()
     response = MockResponse()
     if url.startswith('http://remote'):
         response.content = '{"action":"get","node":{"key":"/service/batman5","dir":true,"nodes":[{"key":"/service/batman5/initialize","value":"postgresql0","modifiedIndex":1582,"createdIndex":1582},{"key":"/service/batman5/leader","value":"postgresql1","expiration":"2015-05-15T09:11:00.037397538Z","ttl":21,"modifiedIndex":20728,"createdIndex":20434},{"key":"/service/batman5/optime","dir":true,"nodes":[{"key":"/service/batman5/optime/leader","value":"2164261704","modifiedIndex":20729,"createdIndex":20729}],"modifiedIndex":20437,"createdIndex":20437},{"key":"/service/batman5/members","dir":true,"nodes":[{"key":"/service/batman5/members/postgresql1","value":"postgres://replicator:rep-pass@127.0.0.1:5434/postgres","expiration":"2015-05-15T09:10:59.949384522Z","ttl":21,"modifiedIndex":20727,"createdIndex":20727},{"key":"/service/batman5/members/postgresql0","value":"postgres://replicator:rep-pass@127.0.0.1:5433/postgres","expiration":"2015-05-15T09:11:09.611860899Z","ttl":30,"modifiedIndex":20730,"createdIndex":20730}],"modifiedIndex":1581,"createdIndex":1581}],"modifiedIndex":1581,"createdIndex":1581}}'
     elif url.startswith('http://other'):
         response.status_code = 404
+    elif url.startswith('http://noleader'):
+        response.content = '{"action":"get","node":{"key":"/service/batman5","dir":true,"nodes":[{"key":"/service/batman5/initialize","value":"postgresql0","modifiedIndex":1582,"createdIndex":1582},{"key":"/service/batman5/leader","value":"postgresql1","expiration":"2015-05-15T09:11:00.037397538Z","ttl":21,"modifiedIndex":20728,"createdIndex":20434},{"key":"/service/batman5/optime","dir":true,"nodes":[{"key":"/service/batman5/optime/leader","value":"2164261704","modifiedIndex":20729,"createdIndex":20729}],"modifiedIndex":20437,"createdIndex":20437},{"key":"/service/batman5/members","dir":true,"nodes":[{"key":"/service/batman5/members/postgresql0","value":"postgres://replicator:rep-pass@127.0.0.1:5433/postgres","expiration":"2015-05-15T09:11:09.611860899Z","ttl":30,"modifiedIndex":20730,"createdIndex":20730}],"modifiedIndex":1581,"createdIndex":1581}],"modifiedIndex":1581,"createdIndex":1581}}'
     return response
 
 
 def requests_put(url, **kwargs):
-    if url.startswith('http://local'):
-        raise Exception()
+    if url.startswith('http://local') or '/optime/leader' in url:
+        raise requests.exceptions.RequestException()
     response = MockResponse()
     response.status_code = 201
+    if url.startswith('http://other'):
+        response.status_code = 404
     return response
 
 
 def requests_delete(url):
     if url.startswith('http://local'):
-        raise Exception()
+        raise requests.exceptions.RequestException()
     response = MockResponse()
     response.status_code = 204
     return response
@@ -65,7 +76,7 @@ class TestEtcd(unittest.TestCase):
         self.assertRaises(Exception, self.etcd.get_client_path, '', 2)
 
     def test_put_client_path(self):
-        self.assertFalse(self.etcd.put_client_path(''))
+        self.assertRaises(EtcdError, self.etcd.put_client_path, '')
 
     def test_delete_client_path(self):
         self.assertFalse(self.etcd.delete_client_path(''))
@@ -77,6 +88,29 @@ class TestEtcd(unittest.TestCase):
         self.assertIsInstance(cluster, Cluster)
         self.etcd.base_client_url = self.etcd.base_client_url.replace('remote', 'other')
         self.etcd.get_cluster()
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('other', 'noleader')
+        self.etcd.get_cluster()
 
     def test_current_leader(self):
         self.assertRaises(CurrentLeaderError, self.etcd.current_leader)
+
+    def test_touch_member(self):
+        self.assertFalse(self.etcd.touch_member('', ''))
+
+    def test_take_leader(self):
+        self.assertFalse(self.etcd.take_leader(''))
+
+    def test_attempt_to_acquire_leader(self):
+        self.assertFalse(self.etcd.attempt_to_acquire_leader(''))
+
+    def test_update_leader(self):
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('local', 'remote')
+        self.assertTrue(self.etcd.update_leader(MockPostgresql()))
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('remote', 'other')
+        self.assertFalse(self.etcd.update_leader(MockPostgresql()))
+
+    def test_race(self):
+        self.assertFalse(self.etcd.race('', ''))
+
+    def test_delete_member(self):
+        self.assertFalse(self.etcd.delete_member(''))
diff --git a/tests/test_ha.py b/tests/test_ha.py
index cabdf7d5..c23244da 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -72,6 +72,7 @@ class TestHa(unittest.TestCase):
         self.p = MockPostgresql()
         self.e = Etcd({'ttl': 30, 'host': 'remotehost', 'scope': 'test'})
         self.ha = Ha(self.p, self.e)
+        self.ha.load_cluster_from_etcd()
         self.ha.cluster = Cluster(None, None, [])
         self.ha.load_cluster_from_etcd = nop
 

From 755f54f26606930a0311705d026a4f11729f6ed1 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sun, 24 May 2015 15:42:41 +0200
Subject: [PATCH 05/12] Improve test coverage for governor.py

---
 governor.py            | 12 +++++++-----
 tests/test_governor.py | 26 +++++++++++++++++++++++---
 2 files changed, 30 insertions(+), 8 deletions(-)

diff --git a/governor.py b/governor.py
index 27102d88..7e62ff8b 100755
--- a/governor.py
+++ b/governor.py
@@ -36,8 +36,8 @@ class Governor:
         self.postgresql = Postgresql(config['postgresql'])
         self.ha = Ha(self.postgresql, self.etcd)
 
-    def touch_member(self):
-        return self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string)
+    def touch_member(self, ttl=None):
+        return self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string, ttl)
 
     def initialize(self):
         # wait for etcd to be available
@@ -72,6 +72,10 @@ class Governor:
 
 
 def main():
+    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+    signal.signal(signal.SIGTERM, sigterm_handler)
+    signal.signal(signal.SIGCHLD, sigchld_handler)
+
     if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
         print('Usage: {} config.yml'.format(sys.argv[0]))
         return
@@ -85,12 +89,10 @@ def main():
         RestApiServer(governor).start()
         governor.run()
     finally:
+        governor.touch_member(300)  # schedule member removal
         governor.postgresql.stop()
         governor.etcd.delete_leader(governor.postgresql.name)
 
 
 if __name__ == '__main__':
-    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
-    signal.signal(signal.SIGTERM, sigterm_handler)
-    signal.signal(signal.SIGCHLD, sigchld_handler)
     main()
diff --git a/tests/test_governor.py b/tests/test_governor.py
index ed7ae32e..844f374e 100644
--- a/tests/test_governor.py
+++ b/tests/test_governor.py
@@ -6,7 +6,7 @@ import sys
 import time
 import yaml
 
-from governor import Governor, main, sigchld_handler
+from governor import Governor, main, sigchld_handler, sigterm_handler
 from test_ha import true, false
 from test_postgresql import Postgresql, os_system, psycopg2_connect
 from test_etcd import requests_get, requests_put, requests_delete
@@ -20,6 +20,10 @@ def os_waitpid(a, b):
     return (0, 0)
 
 
+def time_sleep(_):
+    raise Exception()
+
+
 class TestGovernor(unittest.TestCase):
 
     def __init__(self, method_name='runTest'):
@@ -28,13 +32,13 @@ class TestGovernor(unittest.TestCase):
         super(TestGovernor, self).__init__(method_name)
 
     def set_up(self):
+        self.touched = False
         os.system = os_system
         psycopg2.connect = psycopg2_connect
         requests.get = requests_get
         requests.put = requests_put
         requests.delete = requests_delete
         time.sleep = nop
-        Governor.run = nop
         self.write_pg_hba = Postgresql.write_pg_hba
         self.write_recovery_conf = Postgresql.write_recovery_conf
         Postgresql.write_pg_hba = nop
@@ -44,9 +48,20 @@ class TestGovernor(unittest.TestCase):
         Postgresql.write_pg_hba = self.write_pg_hba
         Postgresql.write_recovery_conf = self.write_recovery_conf
 
+    def test_sigterm_handler(self):
+        self.assertRaises(SystemExit, sigterm_handler, None, None)
+
     def test_governor_main(self):
-        sys.argv = ['governor.py', 'postgres0.yml']
         main()
+        sys.argv = ['governor.py', 'postgres0.yml']
+        time.sleep = time_sleep
+        self.assertRaises(Exception, main)
+
+    def touch_member(self):
+        if not self.touched:
+            self.touched = True
+            return False
+        return True
 
     def test_governor_initialize(self):
         with open('postgres0.yml', 'r') as f:
@@ -60,7 +75,12 @@ class TestGovernor(unittest.TestCase):
             g.etcd.race = false
             g.initialize()
             g.postgresql.data_directory_empty = false
+            g.touch_member = self.touch_member
             g.initialize()
+            g.postgresql.data_directory_empty = true
+            time.sleep = time_sleep
+            g.postgresql.sync_from_leader = false
+            self.assertRaises(Exception, g.initialize)
 
     def test_sigchld_handler(self):
         sigchld_handler(None, None)

From db34e1f4580b28f4d3f7c1b6289cac6ef91b5447 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sun, 24 May 2015 15:57:10 +0200
Subject: [PATCH 06/12] Write get_postgresql_status exception into log file

---
 helpers/api.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/helpers/api.py b/helpers/api.py
index 6269a83c..2b35786c 100644
--- a/helpers/api.py
+++ b/helpers/api.py
@@ -20,6 +20,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
         try:
             response = self.get_postgresql_status()
         except (psycopg2.OperationalError, psycopg2.InterfaceError):
+            logging.exception('get_postgresql_status')
             response = {'running': False}
 
         path = '/master' if self.path == '/' else self.path

From 1096e0434326a36dd34fb89436e1743a0599d676 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sun, 24 May 2015 16:00:31 +0200
Subject: [PATCH 07/12] Bigfix, cursor.closed contains boolean value

---
 helpers/api.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/helpers/api.py b/helpers/api.py
index 2b35786c..f0d47e10 100644
--- a/helpers/api.py
+++ b/helpers/api.py
@@ -65,6 +65,6 @@ class RestApiServer(HTTPServer, Thread):
         self.daemon = True
 
     def cursor(self):
-        if not self._cursor_holder or self._cursor_holder.closed != 0:
+        if not self._cursor_holder or self._cursor_holder.closed:
             self._cursor_holder = self.governor.postgresql.connection().cursor()
         return self._cursor_holder

From faf295dcd095964b2bb2fa783808c69f9a0355b9 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <wmd@mail.ru>
Date: Sun, 24 May 2015 19:26:35 +0200
Subject: [PATCH 08/12] Add initialize key into Cluster

---
 helpers/etcd.py          | 8 +++++---
 tests/test_ha.py         | 2 +-
 tests/test_postgresql.py | 2 +-
 3 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/helpers/etcd.py b/helpers/etcd.py
index 966f881b..6da6a3bc 100644
--- a/helpers/etcd.py
+++ b/helpers/etcd.py
@@ -11,7 +11,7 @@ logger = logging.getLogger(__name__)
 Member = namedtuple('Member', 'hostname,address,ttl')
 
 
-class Cluster(namedtuple('Cluster', 'leader,last_leader_operation,members')):
+class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')):
 
     def is_unlocked(self):
         return not (self.leader and self.leader.hostname)
@@ -87,6 +87,8 @@ class Etcd:
         try:
             response, status_code = self.get_client_path('?recursive=true')
             if status_code == 200:
+                node = self.find_node(response['node'], '/initialize')
+                initialize = True if node else False
                 # get list of members
                 node = self.find_node(response['node'], '/members') or {'nodes': []}
                 members = [Member(n['key'].split('/')[-1], n['value'], n.get('ttl', None)) for n in node['nodes']]
@@ -110,9 +112,9 @@ class Etcd:
                     if not leader:
                         leader = Member(node['value'], None, None)
 
-                return Cluster(leader, last_leader_operation, members)
+                return Cluster(initialize, leader, last_leader_operation, members)
             elif status_code == 404:
-                return Cluster(None, None, [])
+                return Cluster(False, None, None, [])
         except:
             logger.exception('get_cluster')
 
diff --git a/tests/test_ha.py b/tests/test_ha.py
index c23244da..65d7bd6a 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -73,7 +73,7 @@ class TestHa(unittest.TestCase):
         self.e = Etcd({'ttl': 30, 'host': 'remotehost', 'scope': 'test'})
         self.ha = Ha(self.p, self.e)
         self.ha.load_cluster_from_etcd()
-        self.ha.cluster = Cluster(None, None, [])
+        self.ha.cluster = Cluster(False, None, None, [])
         self.ha.load_cluster_from_etcd = nop
 
     def test_start_as_slave(self):
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index e0a21645..2e30b602 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -142,7 +142,7 @@ class TestPostgresql(unittest.TestCase):
         leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', 28)
         me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', 28)
         other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', 28)
-        cluster = Cluster(leader, 0, [me, other, leader])
+        cluster = Cluster(True, leader, 0, [me, other, leader])
         self.assertTrue(self.p.is_healthiest_node(cluster))
         self.p.is_leader = false
         self.assertFalse(self.p.is_healthiest_node(cluster))

From 3e3acc3d3a4928e2a4626551eca53c1d38453fce Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <alexander.kukushkin@zalando.de>
Date: Tue, 26 May 2015 08:59:42 +0200
Subject: [PATCH 09/12] Do not start listening port during test

---
 tests/test_governor.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/tests/test_governor.py b/tests/test_governor.py
index 844f374e..34b46043 100644
--- a/tests/test_governor.py
+++ b/tests/test_governor.py
@@ -11,6 +11,11 @@ from test_ha import true, false
 from test_postgresql import Postgresql, os_system, psycopg2_connect
 from test_etcd import requests_get, requests_put, requests_delete
 
+if sys.hexversion >= 0x03000000:
+    import http.server as BaseHTTPServer
+else:
+    import BaseHTTPServer
+
 
 def nop(*args, **kwargs):
     pass
@@ -43,6 +48,7 @@ class TestGovernor(unittest.TestCase):
         self.write_recovery_conf = Postgresql.write_recovery_conf
         Postgresql.write_pg_hba = nop
         Postgresql.write_recovery_conf = nop
+        BaseHTTPServer.HTTPServer.__init__ = nop
 
     def tear_down(self):
         Postgresql.write_pg_hba = self.write_pg_hba

From cca5bdef8f234c758fb17aace4ceb4efa571da9e Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <alexander.kukushkin@zalando.de>
Date: Tue, 26 May 2015 09:07:39 +0200
Subject: [PATCH 10/12] Configure listen address for restapi via a yaml file

---
 governor.py   | 3 ++-
 postgres0.yml | 2 ++
 postgres1.yml | 2 ++
 3 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/governor.py b/governor.py
index 7e62ff8b..825b4e3b 100755
--- a/governor.py
+++ b/governor.py
@@ -86,7 +86,8 @@ def main():
     governor = Governor(config)
     try:
         governor.initialize()
-        RestApiServer(governor).start()
+        host, port = config['restapi']['listen'].split(':')
+        RestApiServer(governor, host, port).start()
         governor.run()
     finally:
         governor.touch_member(300)  # schedule member removal
diff --git a/postgres0.yml b/postgres0.yml
index 5f4262f9..3f4b4761 100644
--- a/postgres0.yml
+++ b/postgres0.yml
@@ -1,4 +1,6 @@
 loop_wait: 10
+restapi:
+  listen: 127.0.0.1:8008
 etcd:
   scope: batman
   ttl: 30
diff --git a/postgres1.yml b/postgres1.yml
index 31a5dca3..c2d4ce29 100644
--- a/postgres1.yml
+++ b/postgres1.yml
@@ -1,4 +1,6 @@
 loop_wait: 10
+restapi:
+  listen: 127.0.0.1:8009
 etcd:
   scope: batman
   ttl: 30

From 45c4ba85b4111a7682ae01f31669caf2831c4512 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <alexander.kukushkin@zalando.de>
Date: Tue, 26 May 2015 09:17:58 +0200
Subject: [PATCH 11/12] port for http servert must be integer

---
 governor.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/governor.py b/governor.py
index 825b4e3b..4e097920 100755
--- a/governor.py
+++ b/governor.py
@@ -87,7 +87,7 @@ def main():
     try:
         governor.initialize()
         host, port = config['restapi']['listen'].split(':')
-        RestApiServer(governor, host, port).start()
+        RestApiServer(governor, host, int(port)).start()
         governor.run()
     finally:
         governor.touch_member(300)  # schedule member removal

From 518ef68a53698f4f1986e21c55e819cfc5da7ae0 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <alexander.kukushkin@zalando.de>
Date: Tue, 26 May 2015 09:19:20 +0200
Subject: [PATCH 12/12] Use connect_timeout=3s and statement_timeout=2s
 everywhere (even for local connections)

---
 helpers/postgresql.py | 15 ++++++++++-----
 1 file changed, 10 insertions(+), 5 deletions(-)

diff --git a/helpers/postgresql.py b/helpers/postgresql.py
index 720d0775..4726ddc4 100644
--- a/helpers/postgresql.py
+++ b/helpers/postgresql.py
@@ -17,15 +17,19 @@ logger = logging.getLogger(__name__)
 
 def parseurl(url):
     r = urlparse(url)
-    return {
+    ret = {
         'host': r.hostname,
         'port': r.port or 5432,
-        'user': r.username,
-        'password': r.password,
         'database': r.path[1:],
         'fallback_application_name': 'Governor',
-        'connect_timeout': 5,
+        'connect_timeout': 3,
+        'options': '-c statement_timeout=2000',
     }
+    if r.username:
+        ret['user'] = r.username
+    if r.password:
+        ret['password'] = r.password
+    return ret
 
 
 class Postgresql:
@@ -55,7 +59,8 @@ class Postgresql:
 
     def connection(self):
         if not self._connection or self._connection.closed != 0:
-            self._connection = psycopg2.connect('postgres://{}/postgres'.format(self.local_address))
+            r = parseurl('postgres://{}/postgres'.format(self.local_address))
+            self._connection = psycopg2.connect(**r)
             self._connection.autocommit = True
         return self._connection