From d202f72a429bd67008a6ca6e63859b09fda00bcd Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sat, 23 May 2015 20:29:51 +0200 Subject: [PATCH 01/17] 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 8dd99170bfae0cb52ce349c7dab3eecc3ff590e3 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sun, 24 May 2015 15:40:36 +0200 Subject: [PATCH 02/17] 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 Date: Sun, 24 May 2015 15:42:41 +0200 Subject: [PATCH 03/17] 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 Date: Sun, 24 May 2015 15:57:10 +0200 Subject: [PATCH 04/17] 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 Date: Sun, 24 May 2015 16:00:31 +0200 Subject: [PATCH 05/17] 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 Date: Sun, 24 May 2015 19:26:35 +0200 Subject: [PATCH 06/17] 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 Date: Tue, 26 May 2015 08:59:42 +0200 Subject: [PATCH 07/17] 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 Date: Tue, 26 May 2015 09:07:39 +0200 Subject: [PATCH 08/17] 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 Date: Tue, 26 May 2015 09:17:58 +0200 Subject: [PATCH 09/17] 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 Date: Tue, 26 May 2015 09:19:20 +0200 Subject: [PATCH 10/17] 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 From 8cc0aa90b1c426accf9a4120455ba8c1114d3b91 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 26 May 2015 10:39:16 +0200 Subject: [PATCH 11/17] Fix typos when passing connection parameters to psycopg2. --- helpers/postgresql.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 4726ddc4..5138662d 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -169,7 +169,8 @@ class Postgresql: if member.hostname == self.name: continue try: - member_conn = psycopg2.connect(parseurl(member.address)) + r = parseurl(member.address) + member_conn = psycopg2.connect(**r) member_conn.autocommit = True member_cursor = member_conn.cursor() member_cursor.execute( From 56daec0d6c94a48ac1826ab6935efde615980500 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 26 May 2015 15:53:57 +0200 Subject: [PATCH 12/17] Api: Bugfix for current_xlog_location in slaves. --- helpers/api.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/helpers/api.py b/helpers/api.py index f0d47e10..83aa619b 100644 --- a/helpers/api.py +++ b/helpers/api.py @@ -37,7 +37,9 @@ class RestApiHandler(BaseHTTPRequestHandler): 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(), + CASE WHEN pg_is_in_recovery() + THEN null + ELSE pg_current_xlog_location() END, pg_last_xlog_receive_location(), pg_last_xlog_replay_location(), pg_is_in_recovery() AND pg_is_xlog_replay_paused()""") From 9968a66048157ba74d261b71766a8aaeae0d331b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 27 May 2015 11:08:06 +0200 Subject: [PATCH 13/17] helpers/postgresql.py Replace os.system with subprocess.call --- tests/test_governor.py | 7 ++++--- tests/test_postgresql.py | 7 ++++--- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/tests/test_governor.py b/tests/test_governor.py index 34b46043..941d6d45 100644 --- a/tests/test_governor.py +++ b/tests/test_governor.py @@ -1,14 +1,15 @@ import os import psycopg2 -import unittest import requests +import subprocess import sys import time +import unittest import yaml 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_postgresql import Postgresql, subprocess_call, psycopg2_connect from test_etcd import requests_get, requests_put, requests_delete if sys.hexversion >= 0x03000000: @@ -38,7 +39,7 @@ class TestGovernor(unittest.TestCase): def set_up(self): self.touched = False - os.system = os_system + subprocess.call = subprocess_call psycopg2.connect = psycopg2_connect requests.get = requests_get requests.put = requests_put diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 2e30b602..dc6fb512 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,13 +1,14 @@ import os import psycopg2 -import unittest import shutil +import subprocess +import unittest from helpers.etcd import Cluster, Member from helpers.postgresql import Postgresql -def os_system(cmd): +def subprocess_call(cmd, shell=False): return 0 @@ -89,7 +90,7 @@ class TestPostgresql(unittest.TestCase): super(TestPostgresql, self).__init__(method_name) def set_up(self): - os.system = os_system + subprocess.call = subprocess_call self.p = Postgresql({'name': 'test0', 'data_dir': 'data/test0', 'listen': '127.0.0.1, 127.0.0.2:5432', 'connect_address': '127.0.0.2:5432', 'replication': { 'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}}) psycopg2.connect = psycopg2_connect From 56a9f52a1512f89dac142965d5d45035e9371430 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 27 May 2015 11:16:16 +0200 Subject: [PATCH 14/17] Replace os.system with subprocess.call --- helpers/postgresql.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 5138662d..868e9bbf 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -1,6 +1,7 @@ import logging import os import psycopg2 +import subprocess import sys import time @@ -42,7 +43,7 @@ class Postgresql: 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._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir] self.local_address = self.get_local_address() connect_address = config.get('connect_address', None) or self.local_address @@ -98,7 +99,7 @@ class Postgresql: return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == [] def initialize(self): - ret = os.system(self._pg_ctl + ' initdb -o --encoding=UTF8') == 0 + ret = subprocess.call(self._pg_ctl + ['initdb', '-o', '--encoding=UTF8']) == 0 ret and self.write_pg_hba() return ret @@ -112,8 +113,8 @@ class Postgresql: try: os.environ['PGPASSFILE'] = pgpass - return os.system('pg_basebackup -R -D {data_dir} --host={host} --port={port} -U {user}'.format( - data_dir=self.data_dir, **r)) == 0 + return subprocess.call(['pg_basebackup', '-R', '-D', self.data_dir, + '--host=' + r['host'], '--port=' + str(r['port']), '-U', r['user']]) == 0 finally: os.environ.pop('PGPASSFILE') @@ -121,7 +122,7 @@ class Postgresql: return not self.query('SELECT pg_is_in_recovery()').fetchone()[0] def is_running(self): - return os.system(self._pg_ctl + ' status > /dev/null') == 0 + return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0 def start(self): if self.is_running(): @@ -133,18 +134,18 @@ class Postgresql: 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 = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0 ret and self.load_replication_slots() return ret def stop(self): - return os.system(self._pg_ctl + ' stop -m fast') != 0 + return subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) != 0 def reload(self): - return os.system(self._pg_ctl + ' reload') == 0 + return subprocess.call(self._pg_ctl + ['reload']) == 0 def restart(self): - return os.system(self._pg_ctl + ' restart -m fast') == 0 + return subprocess.call(self._pg_ctl + ['restart', '-m', 'fast']) == 0 def server_options(self): options = "--listen_addresses='{}' --port={}".format(self.listen_addresses, self.port) @@ -230,7 +231,7 @@ primary_conninfo = '{}' self.restart() def promote(self): - return os.system(self._pg_ctl + ' promote') == 0 + return subprocess.call(self._pg_ctl + ['promote']) == 0 def demote(self, leader): self.follow_the_leader(leader) From 77911144c75497e31e43015c3a1a8fe27c2727b4 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 27 May 2015 14:44:52 +0200 Subject: [PATCH 15/17] Improve test of governor against real config files --- tests/test_etcd.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 8b3d60b2..8f2280e6 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -28,7 +28,7 @@ def requests_get(url, **kwargs): if url.startswith('http://local'): raise requests.exceptions.RequestException() response = MockResponse() - if url.startswith('http://remote'): + if url.startswith('http://remote') or url.startswith('http://127.0.0.1'): 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 From 5a817fad0dde86a7f0d454d17c532b94253dc7d9 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 27 May 2015 14:46:35 +0200 Subject: [PATCH 16/17] Store in etcd endpoint of restapi In order to have backward compatibility connection url of restapi is passed as application_name parameter in connection url of database. Value in etcd will look like: postgres://user:passwd@host:5432/postgres?application_name=http://host:8080/governor Sure, this is a hack, but I guess it worth it. --- governor.py | 8 +++++--- helpers/api.py | 10 ++++++---- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/governor.py b/governor.py index 4e097920..85cd28e0 100755 --- a/governor.py +++ b/governor.py @@ -35,9 +35,12 @@ class Governor: self.etcd = Etcd(config['etcd']) self.postgresql = Postgresql(config['postgresql']) self.ha = Ha(self.postgresql, self.etcd) + host, port = config['restapi']['listen'].split(':') + self.api = RestApiServer(self, config['restapi']) def touch_member(self, ttl=None): - return self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string, ttl) + connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string + return self.etcd.touch_member(self.postgresql.name, connection_string, ttl) def initialize(self): # wait for etcd to be available @@ -65,6 +68,7 @@ class Governor: self.postgresql.load_replication_slots() def run(self): + self.api.start() while True: self.touch_member() logging.info(self.ha.run_cycle()) @@ -86,8 +90,6 @@ def main(): governor = Governor(config) try: governor.initialize() - host, port = config['restapi']['listen'].split(':') - RestApiServer(governor, host, int(port)).start() governor.run() finally: governor.touch_member(300) # schedule member removal diff --git a/helpers/api.py b/helpers/api.py index 83aa619b..c723e47b 100644 --- a/helpers/api.py +++ b/helpers/api.py @@ -34,7 +34,7 @@ class RestApiHandler(BaseHTTPRequestHandler): def get_postgresql_status(self): if not self.server.governor.postgresql.is_running(): return {'running': False} - cursor = self.server.cursor() + 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(), CASE WHEN pg_is_in_recovery() @@ -59,14 +59,16 @@ class RestApiHandler(BaseHTTPRequestHandler): 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) + def __init__(self, governor, config): + self.connection_string = 'http://{}/governor'.format(config.get('connect_address', None) or config['listen']) + host, port = config['listen'].split(':') + HTTPServer.__init__(self, (host, int(port)), RestApiHandler) Thread.__init__(self, target=self.serve_forever) self.governor = governor self._cursor_holder = None self.daemon = True - def cursor(self): + def _cursor(self): if not self._cursor_holder or self._cursor_holder.closed: self._cursor_holder = self.governor.postgresql.connection().cursor() return self._cursor_holder From bdc4374ae3d6fd6fa6303dc7965cf7aeee3fc4cf Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 27 May 2015 14:51:41 +0200 Subject: [PATCH 17/17] connect address for rest api could be different from listen address --- postgres0.yml | 1 + postgres1.yml | 1 + 2 files changed, 2 insertions(+) diff --git a/postgres0.yml b/postgres0.yml index 3f4b4761..8757538f 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -1,6 +1,7 @@ loop_wait: 10 restapi: listen: 127.0.0.1:8008 + connect_address: 127.0.0.1:8008 etcd: scope: batman ttl: 30 diff --git a/postgres1.yml b/postgres1.yml index c2d4ce29..a0afe20f 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -1,6 +1,7 @@ loop_wait: 10 restapi: listen: 127.0.0.1:8009 + connect_address: 127.0.0.1:8009 etcd: scope: batman ttl: 30