From 275e18e58a3b523b5276f48a34bca2e485d73781 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 10 Jun 2015 14:21:23 +0200 Subject: [PATCH 1/5] Refactor tests --- tests/test_postgresql.py | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 9eba294b..7fae838a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -101,7 +101,9 @@ class TestPostgresql(unittest.TestCase): psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) - self.leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, 28) + self.leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, None, 28) + self.other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, None, 28) + self.me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, None, 28) def tear_down(self): shutil.rmtree('data') @@ -128,13 +130,11 @@ class TestPostgresql(unittest.TestCase): self.p.follow_the_leader(None) self.p.demote(self.leader) self.p.follow_the_leader(self.leader) - self.p.follow_the_leader(Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, 28)) + self.p.follow_the_leader(self.other) def test_create_replication_slots(self): self.p.start() - me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, 28) - other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, 28) - cluster = Cluster(True, self.leader, 0, [me, other, self.leader]) + cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader]) self.p.create_replication_slots(cluster) def test_query(self): @@ -148,10 +148,7 @@ class TestPostgresql(unittest.TestCase): self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla') def test_is_healthiest_node(self): - leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, 28) - me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, 28) - other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, 28) - cluster = Cluster(True, leader, 0, [me, other, leader]) + cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader]) self.assertTrue(self.p.is_healthiest_node(cluster)) self.p.is_leader = false self.assertFalse(self.p.is_healthiest_node(cluster)) From 44e06b4cffa61826461e46fb017f6ccaa07f9c37 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 10 Jun 2015 14:22:11 +0200 Subject: [PATCH 2/5] Add expiration field into Member --- helpers/etcd.py | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 143a8ff1..3a222a3b 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -19,18 +19,16 @@ else: logger = logging.getLogger(__name__) -class Member(namedtuple('Member', 'hostname,conn_url,api_url,ttl')): +class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): @staticmethod def fromNode(node): scheme, netloc, path, params, query, fragment = urlparse(node['value']) conn_url = urlunparse((scheme, netloc, path, params, '', fragment)) - api_url = None - for name, value in parse_qsl(query): - if name == 'application_name' and value: - api_url = value - break - return Member(node['key'].split('/')[-1], conn_url, api_url, node.get('ttl', None)) + api_url = ([v for n, v in parse_qsl(query) if n == 'application_name'] or [None])[0] + expiration = node.get('expiration', None) + ttl = node.get('ttl', None) + return Member(node['key'].split('/')[-1], conn_url, api_url, expiration, ttl) class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): @@ -271,7 +269,7 @@ class Etcd: leader = m break if not leader: - leader = Member(node['value'], None, None, None) + leader = Member(node['value'], None, None, None, None) return Cluster(initialize, leader, last_leader_operation, members) elif status_code == 404: From 744026b4bb44eaeb93f7ba950c42ef9aff8f5b0b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 23 Jun 2015 14:02:32 +0200 Subject: [PATCH 3/5] Do not update member TTL when it is far from being expired --- governor.py | 8 +++++++- helpers/etcd.py | 5 ++++- helpers/utils.py | 29 +++++++++++++++++++++++++++++ tests/test_etcd.py | 15 ++++++++++++++- tests/test_governor.py | 9 +++++++++ 5 files changed, 63 insertions(+), 3 deletions(-) diff --git a/governor.py b/governor.py index fe2ecd8a..ba48fefd 100755 --- a/governor.py +++ b/governor.py @@ -22,9 +22,15 @@ class Governor: host, port = config['restapi']['listen'].split(':') self.api = RestApiServer(self, config['restapi']) self.next_run = time.time() + self.shutdown_member_ttl = 300 def touch_member(self, ttl=None): connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string + if self.ha.cluster: + for m in self.ha.cluster.members: + # Do not update member TTL when it is far from being expired + if m.hostname == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: + return True return self.etcd.touch_member(self.postgresql.name, connection_string, ttl) def initialize(self): @@ -91,7 +97,7 @@ def main(): except KeyboardInterrupt: pass finally: - governor.touch_member(300) # schedule member removal + governor.touch_member(governor.shutdown_member_ttl) # schedule member removal governor.postgresql.stop() governor.etcd.delete_leader(governor.postgresql.name) diff --git a/helpers/etcd.py b/helpers/etcd.py index 3a222a3b..f63cc364 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -8,7 +8,7 @@ from collections import namedtuple from dns.exception import DNSException from dns import resolver from helpers.errors import CurrentLeaderError, EtcdError, EtcdConnectionFailed -from helpers.utils import sleep +from helpers.utils import calculate_ttl, sleep from requests.exceptions import RequestException if sys.hexversion >= 0x03000000: @@ -30,6 +30,9 @@ class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): ttl = node.get('ttl', None) return Member(node['key'].split('/')[-1], conn_url, api_url, expiration, ttl) + def real_ttl(self): + return calculate_ttl(self.expiration) + class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): diff --git a/helpers/utils.py b/helpers/utils.py index e08dd657..c12f389a 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -1,10 +1,39 @@ +import datetime import os +import re import signal import sys import time received_sigchld = False +_DATE_TIME_RE = re.compile(r'''^ +(?P\d{4})\-(?P\d{2})\-(?P\d{2}) # date +T +(?P\d{2}):(?P\d{2}):(?P\d{2})\.(?P\d{6}) # time +\d*Z$''', re.X) + + +def parse_datetime(time_str): + """ + >>> parse_datetime('2015-06-10T12:56:30.552539016Z') + datetime.datetime(2015, 6, 10, 12, 56, 30, 552539) + >>> parse_datetime('2015-06-10 12:56:30.552539016Z') + """ + m = _DATE_TIME_RE.match(time_str) + if not m: + return None + p = dict((n, int(m.group(n))) for n in 'year month day hour minute second microsecond'.split(' ')) + return datetime.datetime(**p) + + +def calculate_ttl(expiration): + expiration = parse_datetime(expiration) + if not expiration: + return None + now = datetime.datetime.utcnow() + return int((expiration - now).total_seconds()) + def lsn_to_bytes(value): """ diff --git a/tests/test_etcd.py b/tests/test_etcd.py index c2e07c55..b9e88df7 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -1,3 +1,4 @@ +import datetime import dns.resolver import json import requests @@ -7,7 +8,7 @@ import unittest from dns.exception import DNSException from helpers.errors import EtcdError, CurrentLeaderError, EtcdConnectionFailed -from helpers.etcd import Client, Cluster, Etcd +from helpers.etcd import Client, Cluster, Etcd, Member class MockResponse: @@ -94,6 +95,18 @@ def socket_getaddrinfo(*args): raise socket.error() +class TestMember(unittest.TestCase): + + def __init__(self, method_name='runTest'): + super(TestMember, self).__init__(method_name) + + def test_real_ttl(self): + now = datetime.datetime.utcnow() + member = Member('a', 'b', 'c', (now + datetime.timedelta(seconds=2)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None) + self.assertLess(member.real_ttl(), 2) + self.assertIsNone(Member('a', 'b', 'c', '', None).real_ttl()) + + class TestClient(unittest.TestCase): def __init__(self, method_name='runTest'): diff --git a/tests/test_governor.py b/tests/test_governor.py index 9ecb88f1..b84637a7 100644 --- a/tests/test_governor.py +++ b/tests/test_governor.py @@ -1,3 +1,4 @@ +import datetime import psycopg2 import requests import subprocess @@ -7,6 +8,7 @@ import unittest import yaml from governor import Governor, main +from helpers.etcd import Cluster, Member from test_ha import true, false from test_postgresql import Postgresql, subprocess_call, psycopg2_connect from test_etcd import requests_get, requests_put, requests_delete @@ -67,6 +69,13 @@ class TestGovernor(unittest.TestCase): return False return True + def test_touch_member(self): + now = datetime.datetime.utcnow() + member = Member(self.g.postgresql.name, 'b', 'c', (now + datetime.timedelta( + seconds=self.g.shutdown_member_ttl + 10)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None) + self.g.ha.cluster = Cluster(True, member, 0, [member]) + self.g.touch_member() + def test_governor_initialize(self): self.g.etcd.client._base_uri = 'http://remote' self.g.postgresql.data_directory_empty = true From b7c73fdba8d2de0113c3a5aa8021d18ea925782a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 23 Jun 2015 14:07:00 +0200 Subject: [PATCH 4/5] Rename hostname to name in a Member --- governor.py | 2 +- helpers/etcd.py | 6 +++--- helpers/ha.py | 2 +- helpers/postgresql.py | 6 +++--- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/governor.py b/governor.py index ba48fefd..d8a1e620 100755 --- a/governor.py +++ b/governor.py @@ -29,7 +29,7 @@ class Governor: if self.ha.cluster: for m in self.ha.cluster.members: # Do not update member TTL when it is far from being expired - if m.hostname == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: + if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: return True return self.etcd.touch_member(self.postgresql.name, connection_string, ttl) diff --git a/helpers/etcd.py b/helpers/etcd.py index f63cc364..0a1f7cb7 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -19,7 +19,7 @@ else: logger = logging.getLogger(__name__) -class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): +class Member(namedtuple('Member', 'name,conn_url,api_url,expiration,ttl')): @staticmethod def fromNode(node): @@ -37,7 +37,7 @@ class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): def is_unlocked(self): - return not (self.leader and self.leader.hostname) + return not (self.leader and self.leader.name) class Client: @@ -268,7 +268,7 @@ class Etcd: node = self.find_node(response['node'], '/leader') if node: for m in members: - if m.hostname == node['value']: + if m.name == node['value']: leader = m break if not leader: diff --git a/helpers/ha.py b/helpers/ha.py index 9617228f..0453686e 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -23,7 +23,7 @@ class Ha: return self.etcd.update_leader(self.state_handler) def has_lock(self): - lock_owner = self.cluster.leader and self.cluster.leader.hostname + lock_owner = self.cluster.leader and self.cluster.leader.name logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name) return lock_owner == self.state_handler.name diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 3d2cd183..861074e8 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -182,7 +182,7 @@ class Postgresql: return False for member in cluster.members: - if member.hostname == self.name: + if member.name == self.name: continue try: r = parseurl(member.conn_url) @@ -195,7 +195,7 @@ class Postgresql: row = member_cursor.fetchone() member_cursor.close() member_conn.close() - logger.error([self.name, member.hostname, row]) + logger.error([self.name, member.name, row]) if not row[0] or row[1] < 0: return False except psycopg2.Error: @@ -269,7 +269,7 @@ primary_conninfo = '{}' self.members = [r[0] for r in cursor] def create_replication_slots(self, cluster): - members = [m.hostname for m in cluster.members if m.hostname != self.name] + members = [m.name for m in cluster.members if m.name != self.name] # drop unused slots for slot in set(self.members) - set(members): self.query("""SELECT pg_drop_replication_slot(%s) From f7be52f63af7a0d4c21a020726d4958e5357c989 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 23 Jun 2015 17:15:37 +0200 Subject: [PATCH 5/5] real_ttl method will return -1 when expiration is not defined --- helpers/etcd.py | 2 +- helpers/utils.py | 6 ++++++ tests/test_etcd.py | 2 +- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 0a1f7cb7..495046f1 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -31,7 +31,7 @@ class Member(namedtuple('Member', 'name,conn_url,api_url,expiration,ttl')): return Member(node['key'].split('/')[-1], conn_url, api_url, expiration, ttl) def real_ttl(self): - return calculate_ttl(self.expiration) + return calculate_ttl(self.expiration) or -1 class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): diff --git a/helpers/utils.py b/helpers/utils.py index c12f389a..5f09490e 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -28,6 +28,12 @@ def parse_datetime(time_str): def calculate_ttl(expiration): + """ + >>> calculate_ttl(None) + >>> calculate_ttl('2015-06-10 12:56:30.552539016Z') + """ + if not expiration: + return None expiration = parse_datetime(expiration) if not expiration: return None diff --git a/tests/test_etcd.py b/tests/test_etcd.py index b9e88df7..09dac847 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -104,7 +104,7 @@ class TestMember(unittest.TestCase): now = datetime.datetime.utcnow() member = Member('a', 'b', 'c', (now + datetime.timedelta(seconds=2)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None) self.assertLess(member.real_ttl(), 2) - self.assertIsNone(Member('a', 'b', 'c', '', None).real_ttl()) + self.assertEquals(Member('a', 'b', 'c', '', None).real_ttl(), -1) class TestClient(unittest.TestCase):