From 3b1efff53e0db0e05b8bb7a2a4443994631293a2 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 27 Aug 2015 10:53:22 +0200 Subject: [PATCH 01/36] Refactor `Cluster` object `Cluster.leader` is not reference to `Member` anymore, but to `Leader` `Leader` class contains field `index` (update index). This field is very useful for watching for events which changing leader key. Also `Leader` contains `member` field, which should reference real member. --- helpers/dcs.py | 15 ++++++++++++--- helpers/etcd.py | 7 ++++--- helpers/ha.py | 2 +- helpers/postgresql.py | 8 ++++---- helpers/zookeeper.py | 27 ++++++++++++--------------- tests/test_etcd.py | 12 ++++++++---- tests/test_patroni.py | 12 ++++++++---- tests/test_postgresql.py | 11 ++++++----- tests/test_zookeeper.py | 12 +++++++++--- 9 files changed, 64 insertions(+), 42 deletions(-) diff --git a/helpers/dcs.py b/helpers/dcs.py index c7140c22..f5984fd9 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -39,7 +39,7 @@ class DCSError(Exception): class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')): """Immutable object (namedtuple) which represents single member of PostgreSQL cluster. Consists of the following fields: - :param index: modification index of a given member key in DCS + :param index: modification index of a given member key in a Configuration Store :param name: name of PostgreSQL cluster member :param conn_url: connection string containing host, user and password which could be used to access this member. :param api_url: REST API url of patroni instance @@ -50,17 +50,26 @@ class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')) return calculate_ttl(self.expiration) or -1 +class Leader(namedtuple('Leader', 'index,expiration,ttl,member')): + """Immutable object (namedtuple) which represents leader key. + Consists of the following fields: + :param index: modification index of a leader key in a Configuration Store + :param expiration: expiration time of the leader key + :param ttl: ttl of the leader key + :param member: reference to a `Member` object which represents current leader (see `Cluster.members`)""" + + class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): """Immutable object (namedtuple) which represents PostgreSQL cluster. Consists of the following fields: :param initialize: boolean, shows whether this cluster has initialization key stored in DC or not. - :param leader: `Member` object which represents current leader of the cluster + :param leader: `Leader` object which represents current leader of the cluster :param last_leader_operation: int or long object containing position of last known leader operation. This value is stored in `/optime/leader` key :param members: list of Member object, all PostgreSQL cluster members including leader""" def is_unlocked(self): - return not (self.leader and self.leader.name) + return not (self.leader and self.leader.member.name) class AbstractDCS: diff --git a/helpers/etcd.py b/helpers/etcd.py index add736c5..f767175b 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -8,7 +8,7 @@ import socket from dns.exception import DNSException from dns import resolver -from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string +from helpers.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string from helpers.utils import sleep from requests.exceptions import RequestException @@ -170,8 +170,9 @@ class Etcd(AbstractDCS): # get leader leader = nodes.get('leader', None) if leader: - leader = Member(-1, leader.value, None, None, None, None) - leader = ([m for m in members if m.name == leader.name] or [leader])[0] + member = Member(-1, leader.value, None, None, None, None) + member = ([m for m in members if m.name == leader.value] or [member])[0] + leader = Leader(leader.modifiedIndex, leader.expiration, leader.ttl, member) return Cluster(initialize, leader, last_leader_operation, members) except etcd.EtcdKeyNotFound: diff --git a/helpers/ha.py b/helpers/ha.py index 8e283c55..ff67f0e0 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -31,7 +31,7 @@ class Ha: return self.dcs.update_leader(self.state_handler) def has_lock(self): - lock_owner = self.cluster.leader and self.cluster.leader.name + lock_owner = self.cluster.leader and self.cluster.leader.member.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 ce8ca18f..ca4ea9c0 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -128,7 +128,7 @@ class Postgresql: os.path.exists(self.trigger_file) and os.unlink(self.trigger_file) def sync_from_leader(self, leader): - r = parseurl(leader.conn_url) + r = parseurl(leader.member.conn_url) pgpass = 'pgpass' with open(pgpass, 'w') as f: @@ -288,7 +288,7 @@ class Postgresql: if not os.path.isfile(self.recovery_conf): return False - pattern = leader and leader.conn_url and self.primary_conninfo(leader.conn_url) + pattern = leader and leader.member.conn_url and self.primary_conninfo(leader.member.conn_url) with open(self.recovery_conf, 'r') as f: for line in f: @@ -304,11 +304,11 @@ class Postgresql: f.write("""standby_mode = 'on' recovery_target_timeline = 'latest' """) - if leader and leader.conn_url: + if leader and leader.member.conn_url: f.write(""" primary_slot_name = '{}' primary_conninfo = '{}' -""".format(self.name, self.primary_conninfo(leader.conn_url))) +""".format(self.name, self.primary_conninfo(leader.member.conn_url))) for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) diff --git a/helpers/zookeeper.py b/helpers/zookeeper.py index cb2918cd..56e755a0 100644 --- a/helpers/zookeeper.py +++ b/helpers/zookeeper.py @@ -3,7 +3,7 @@ import random import requests import time -from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string +from helpers.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string from helpers.utils import sleep from kazoo.client import KazooClient, KazooState from kazoo.exceptions import NoNodeError, NodeExistsError @@ -134,21 +134,18 @@ class ZooKeeper(AbstractDCS): leader = self.get_node('/leader', self.cluster_watcher) self.members = self.load_members() if leader: - if leader[0] == self._name: - client_id = self.client.client_id - if client_id is not None and client_id[0] != leader[1].ephemeralOwner: - logger.info('I am leader but not owner of the session. Removing leader node') - self.client.delete(self.client_path('/leader')) - leader = None + client_id = self.client.client_id + if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner: + logger.info('I am leader but not owner of the session. Removing leader node') + self.client.delete(self.client_path('/leader')) + leader = None if leader: - for member in self.members: - if member.name == leader[0]: - leader = member - self.fetch_cluster = False - break - if not isinstance(leader, Member): - leader = Member(-1, leader, None, None, None, None) + member = Member(-1, leader[0], None, None, None, None) + member = ([m for m in self.members if m.name == leader[0]] or [member])[0] + leader = Leader(leader[1].mzxid, None, None, member) + self.fetch_cluster = member.index == -1 + self.leader = leader if self.fetch_cluster: last_leader_operation = self.get_node('/optime/leader') @@ -220,7 +217,7 @@ class ZooKeeper(AbstractDCS): return True def delete_leader(self): - if isinstance(self.leader, Member) and self.leader.name == self._name: + if isinstance(self.leader, Leader) and self.leader.member.name == self._name: self.client.delete(self.client_path('/leader')) def sleep(self, timeout): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 38692c52..52de036e 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -8,7 +8,7 @@ import time import unittest from dns.exception import DNSException -from helpers.dcs import Cluster, DCSError, Member +from helpers.dcs import Cluster, DCSError, Leader, Member from helpers.etcd import Client, Etcd from mock import Mock, patch @@ -107,8 +107,12 @@ def time_sleep(_): pass +class SleepException(Exception): + pass + + def time_sleep_exception(_): - raise Exception() + raise SleepException() class MockSRV: @@ -204,7 +208,7 @@ class TestEtcd(unittest.TestCase): time.sleep = time_sleep_exception with patch.object(etcd.Client, 'machines') as mock_machines: mock_machines.__get__ = Mock(side_effect=etcd.EtcdException) - self.assertRaises(Exception, self.etcd.get_etcd_client, {'discovery_srv': 'test'}) + self.assertRaises(SleepException, self.etcd.get_etcd_client, {'discovery_srv': 'test'}) def test_get_cluster(self): self.assertIsInstance(self.etcd.get_cluster(), Cluster) @@ -214,7 +218,7 @@ class TestEtcd(unittest.TestCase): self.assertIsNone(cluster.leader) def test_current_leader(self): - self.assertIsInstance(self.etcd.current_leader(), Member) + self.assertIsInstance(self.etcd.current_leader(), Leader) self.etcd._base_path = '/service/noleader' self.assertIsNone(self.etcd.current_leader()) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 67ac1ffd..575c7cf3 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -24,8 +24,12 @@ def nop(*args, **kwargs): pass +class SleepException(Exception): + pass + + def time_sleep(*args): - raise Exception() + raise SleepException() class Mock_BaseServer__is_shut_down: @@ -90,7 +94,7 @@ class TestPatroni(unittest.TestCase): Etcd.delete_leader = nop - self.assertRaises(Exception, main) + self.assertRaises(SleepException, main) Patroni.run = run Patroni.touch_member = touch_member @@ -100,10 +104,10 @@ class TestPatroni(unittest.TestCase): self.p.touch_member = self.touch_member self.p.ha.state_handler.sync_replication_slots = time_sleep self.p.ha.dcs.client.read = etcd_read - self.assertRaises(Exception, self.p.run) + self.assertRaises(SleepException, self.p.run) self.p.ha.state_handler.is_leader = lambda: False self.p.api.start = nop - self.assertRaises(Exception, self.p.run) + self.assertRaises(SleepException, self.p.run) def touch_member(self, ttl=None): if not self.touched: diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 040d1c66..08ba0b33 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -4,7 +4,7 @@ import shutil import subprocess import unittest -from helpers.dcs import Cluster, Member +from helpers.dcs import Cluster, Leader, Member from helpers.postgresql import Postgresql @@ -123,7 +123,8 @@ 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(0, 'leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, None, 28) + self.leadermem = Member(0, 'leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, None, 28) + self.leader = Leader(-1, None, 28, self.leadermem) self.other = Member(0, 'test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, None, 28) self.me = Member(0, 'test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, None, 28) @@ -156,7 +157,7 @@ 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(self.other) + self.p.follow_the_leader(Leader(-1, None, 28, self.other)) def test_create_connection_users(self): cfg = self.p.config @@ -166,7 +167,7 @@ class TestPostgresql(unittest.TestCase): def test_create_replication_slots(self): self.p.start() - cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader]) + cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem]) self.p.create_replication_slots(cluster) def test_query(self): @@ -180,7 +181,7 @@ class TestPostgresql(unittest.TestCase): self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla') def test_is_healthiest_node(self): - cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader]) + cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem]) self.assertTrue(self.p.is_healthiest_node(cluster)) self.p.is_leader = false self.assertFalse(self.p.is_healthiest_node(cluster)) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 53f5ab82..54adea5e 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -2,6 +2,7 @@ import helpers.zookeeper import requests import unittest +from helpers.dcs import Leader from helpers.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError from kazoo.client import KazooState from kazoo.exceptions import NoNodeError, NodeExistsError @@ -30,6 +31,10 @@ class MockEventHandler: return MockEvent() +class SleepException(Exception): + pass + + class MockKazooClient: def __init__(self, **kwargs): @@ -94,7 +99,7 @@ class MockKazooClient: def exhibitor_sleep(_): - raise Exception + raise SleepException class TestExhibitorEnsembleProvider(unittest.TestCase): @@ -108,7 +113,7 @@ class TestExhibitorEnsembleProvider(unittest.TestCase): helpers.zookeeper.sleep = exhibitor_sleep def test_init(self): - self.assertRaises(Exception, ExhibitorEnsembleProvider, ['localhost'], 8181) + self.assertRaises(SleepException, ExhibitorEnsembleProvider, ['localhost'], 8181) class TestZooKeeper(unittest.TestCase): @@ -136,7 +141,8 @@ class TestZooKeeper(unittest.TestCase): def test_get_cluster(self): self.assertRaises(ZooKeeperError, self.zk.get_cluster) self.zk.exhibitor.poll = lambda: True - self.zk.get_cluster() + cluster = self.zk.get_cluster() + self.assertIsInstance(cluster.leader, Leader) self.zk.touch_member('foo') self.zk.delete_leader() From fab321c6b07c7cd567904fa328f88bf5a1786201 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Sep 2015 09:54:57 +0200 Subject: [PATCH 02/36] Watch for changes of leader key --- helpers/etcd.py | 34 +++++++++++++++++++++++++++++++++- tests/test_etcd.py | 24 ++++++++++++++++++++++++ tests/test_patroni.py | 3 ++- 3 files changed, 59 insertions(+), 2 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index f767175b..ac83fce8 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -5,6 +5,8 @@ import os import random import requests import socket +import time +import urllib3 from dns.exception import DNSException from dns import resolver @@ -59,6 +61,11 @@ class Client(etcd.Client): logger.exception('Can not resolve SRV for %s', host) return [] + # try to workarond bug in python-etcd: https://github.com/jplana/python-etcd/issues/81 + def _result_from_response(self, response): + response.data.decode('utf-8') + return super(Client, self)._result_from_response(response) + def _get_machines_cache_from_srv(self, discovery_srv): """Fetch list of etcd-cluster member by resolving _etcd-server._tcp. SRV record. This record should contain list of host and peer ports which could be used to run @@ -136,6 +143,7 @@ class Etcd(AbstractDCS): self.ttl = config['ttl'] self.member_ttl = config.get('member_ttl', 3600) self.client = self.get_etcd_client(config) + self.cluster = None def get_etcd_client(self, config): client = None @@ -174,12 +182,14 @@ class Etcd(AbstractDCS): member = ([m for m in members if m.name == leader.value] or [member])[0] leader = Leader(leader.modifiedIndex, leader.expiration, leader.ttl, member) - return Cluster(initialize, leader, last_leader_operation, members) + self.cluster = Cluster(initialize, leader, last_leader_operation, members) + return self.cluster except etcd.EtcdKeyNotFound: return Cluster(False, None, None, []) except: logger.exception('get_cluster') + self.cluster = None raise EtcdError('Etcd is not responding properly') @catch_etcd_errors @@ -213,3 +223,25 @@ class Etcd(AbstractDCS): @catch_etcd_errors def delete_leader(self): return self.client.delete(self.client_path('/leader'), prevValue=self._name) + + def sleep(self, timeout): + # watch on leader key changes if it is defined and current node is not lock owner + if self.cluster and self.cluster.leader and self.cluster.leader.member.name != self._name: + end_time = time.time() + timeout + index = self.cluster.leader.index + + while index and timeout >= 1: # when timeout is too small urllib3 doesn't have enough time to connect + try: + res = self.client.watch(self.client_path('/leader'), index=index + 1, timeout=timeout) + if res.action not in ['set', 'compareAndSwap'] or res.value != self.cluster.leader.member.name: + return + index = res.modifiedIndex + except urllib3.exceptions.TimeoutError: + self.client.http.clear() + return + except etcd.EtcdException: + index = None + + timeout = end_time - time.time() + + timeout > 0 and super(Etcd, self).sleep(timeout) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 52de036e..8de4ca46 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -3,6 +3,7 @@ import dns.resolver import etcd import json import requests +import urllib3 import socket import time import unittest @@ -61,6 +62,20 @@ def requests_get(url, **kwargs): return response +def etcd_watch(key, index=None, timeout=None, recursive=None): + print ('watch', key, index, timeout) + if timeout == 1: + raise urllib3.exceptions.TimeoutError + elif timeout == 5: + return etcd.EtcdResult('delete', {}) + elif timeout == 10: + raise etcd.EtcdException + elif index == 20729: + return etcd.EtcdResult('set', {'value': 'postgresql1', 'modifiedIndex': index + 1}) + elif index == 20731: + return etcd.EtcdResult('set', {'value': 'postgresql2', 'modifiedIndex': index + 1}) + + def etcd_write(key, value, **kwargs): if key == '/service/test/leader': if kwargs.get('prevValue', None) == 'foo' or not kwargs.get('prevExist', True): @@ -237,3 +252,12 @@ class TestEtcd(unittest.TestCase): def test_delete_leader(self): self.etcd.client.delete = etcd_delete self.assertFalse(self.etcd.delete_leader()) + + def test_sleep(self): + self.etcd.client.watch = etcd_watch + self.etcd.sleep(100) + self.etcd.get_cluster() + self.etcd.sleep(1) + self.etcd.sleep(5) + self.etcd.sleep(10) + self.etcd.sleep(100) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 575c7cf3..c0ea2907 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -14,7 +14,7 @@ from helpers.zookeeper import ZooKeeper from mock import Mock, patch from patroni import Patroni, main from six.moves import BaseHTTPServer -from test_etcd import Client, etcd_read, etcd_write +from test_etcd import Client, etcd_read, etcd_write, etcd_watch from test_ha import true, false from test_postgresql import Postgresql, subprocess_call, psycopg2_connect from test_zookeeper import MockKazooClient @@ -104,6 +104,7 @@ class TestPatroni(unittest.TestCase): self.p.touch_member = self.touch_member self.p.ha.state_handler.sync_replication_slots = time_sleep self.p.ha.dcs.client.read = etcd_read + self.p.ha.dcs.sleep = time_sleep self.assertRaises(SleepException, self.p.run) self.p.ha.state_handler.is_leader = lambda: False self.p.api.start = nop From 10c95a23e4ff11cf1871bbb56a66b986a706878a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Sep 2015 09:59:37 +0200 Subject: [PATCH 03/36] Rename sleep to watch in a AbstractDCS This method suppose to watch for changes of leader key if current node is not leader and also it could watch for changes in a members list if current conde is the leader. --- helpers/dcs.py | 2 +- helpers/etcd.py | 4 ++-- helpers/zookeeper.py | 2 +- tests/test_etcd.py | 12 ++++++------ tests/test_zookeeper.py | 4 ++-- 5 files changed, 12 insertions(+), 12 deletions(-) diff --git a/helpers/dcs.py b/helpers/dcs.py index f5984fd9..63515e81 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -153,5 +153,5 @@ class AbstractDCS: """Voluntarily remove leader key from DCS This method should remove leader key if current instance is the leader""" - def sleep(self, timeout): + def watch(self, timeout): sleep(timeout) diff --git a/helpers/etcd.py b/helpers/etcd.py index ac83fce8..6bea59e7 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -224,7 +224,7 @@ class Etcd(AbstractDCS): def delete_leader(self): return self.client.delete(self.client_path('/leader'), prevValue=self._name) - def sleep(self, timeout): + def watch(self, timeout): # watch on leader key changes if it is defined and current node is not lock owner if self.cluster and self.cluster.leader and self.cluster.leader.member.name != self._name: end_time = time.time() + timeout @@ -244,4 +244,4 @@ class Etcd(AbstractDCS): timeout = end_time - time.time() - timeout > 0 and super(Etcd, self).sleep(timeout) + timeout > 0 and super(Etcd, self).watch(timeout) diff --git a/helpers/zookeeper.py b/helpers/zookeeper.py index 56e755a0..bf8b281a 100644 --- a/helpers/zookeeper.py +++ b/helpers/zookeeper.py @@ -220,7 +220,7 @@ class ZooKeeper(AbstractDCS): if isinstance(self.leader, Leader) and self.leader.member.name == self._name: self.client.delete(self.client_path('/leader')) - def sleep(self, timeout): + def watch(self, timeout): self.cluster_event.wait(timeout) if self.cluster_event.isSet(): self.fetch_cluster = True diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 8de4ca46..6754a68a 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -253,11 +253,11 @@ class TestEtcd(unittest.TestCase): self.etcd.client.delete = etcd_delete self.assertFalse(self.etcd.delete_leader()) - def test_sleep(self): + def test_watch(self): self.etcd.client.watch = etcd_watch - self.etcd.sleep(100) + self.etcd.watch(100) self.etcd.get_cluster() - self.etcd.sleep(1) - self.etcd.sleep(5) - self.etcd.sleep(10) - self.etcd.sleep(100) + self.etcd.watch(1) + self.etcd.watch(5) + self.etcd.watch(10) + self.etcd.watch(100) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 54adea5e..01100c4e 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -164,5 +164,5 @@ class TestZooKeeper(unittest.TestCase): self.zk.last_leader_operation = -1 self.assertTrue(self.zk.update_leader(MockPostgresql())) - def test_sleep(self): - self.zk.sleep(0) + def test_watch(self): + self.zk.watch(0) From 5a0634d7b0705cfc0ac1d18265e22b96dbecbe9b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Sep 2015 16:37:03 +0200 Subject: [PATCH 04/36] Create name property in a Cluster object It returns self.member.name and simplifies later usage in code. --- helpers/dcs.py | 9 +++++++-- helpers/etcd.py | 7 ++++--- helpers/ha.py | 2 +- helpers/zookeeper.py | 2 +- 4 files changed, 13 insertions(+), 7 deletions(-) diff --git a/helpers/dcs.py b/helpers/dcs.py index 63515e81..22c47e26 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -58,6 +58,10 @@ class Leader(namedtuple('Leader', 'index,expiration,ttl,member')): :param ttl: ttl of the leader key :param member: reference to a `Member` object which represents current leader (see `Cluster.members`)""" + @property + def name(self): + return self.member.name + class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): """Immutable object (namedtuple) which represents PostgreSQL cluster. @@ -69,7 +73,7 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem :param members: list of Member object, all PostgreSQL cluster members including leader""" def is_unlocked(self): - return not (self.leader and self.leader.member.name) + return not (self.leader and self.leader.name) class AbstractDCS: @@ -83,7 +87,8 @@ class AbstractDCS: i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc... """ self._name = name - self._base_path = '/service/' + config['scope'] + self._scope = config['scope'] + self._base_path = '/service/' + self._scope def client_path(self, path): return self._base_path + path diff --git a/helpers/etcd.py b/helpers/etcd.py index 6bea59e7..d6739e2d 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -185,7 +185,8 @@ class Etcd(AbstractDCS): self.cluster = Cluster(initialize, leader, last_leader_operation, members) return self.cluster except etcd.EtcdKeyNotFound: - return Cluster(False, None, None, []) + self.cluster = Cluster(False, None, None, []) + return self.cluster except: logger.exception('get_cluster') @@ -226,14 +227,14 @@ class Etcd(AbstractDCS): def watch(self, timeout): # watch on leader key changes if it is defined and current node is not lock owner - if self.cluster and self.cluster.leader and self.cluster.leader.member.name != self._name: + if self.cluster and self.cluster.leader and self.cluster.leader.name != self._name: end_time = time.time() + timeout index = self.cluster.leader.index while index and timeout >= 1: # when timeout is too small urllib3 doesn't have enough time to connect try: res = self.client.watch(self.client_path('/leader'), index=index + 1, timeout=timeout) - if res.action not in ['set', 'compareAndSwap'] or res.value != self.cluster.leader.member.name: + if res.action not in ['set', 'compareAndSwap'] or res.value != self.cluster.leader.name: return index = res.modifiedIndex except urllib3.exceptions.TimeoutError: diff --git a/helpers/ha.py b/helpers/ha.py index ff67f0e0..8e283c55 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -31,7 +31,7 @@ class Ha: return self.dcs.update_leader(self.state_handler) def has_lock(self): - lock_owner = self.cluster.leader and self.cluster.leader.member.name + 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/zookeeper.py b/helpers/zookeeper.py index bf8b281a..d1ccafa2 100644 --- a/helpers/zookeeper.py +++ b/helpers/zookeeper.py @@ -217,7 +217,7 @@ class ZooKeeper(AbstractDCS): return True def delete_leader(self): - if isinstance(self.leader, Leader) and self.leader.member.name == self._name: + if isinstance(self.leader, Leader) and self.leader.name == self._name: self.client.delete(self.client_path('/leader')) def watch(self, timeout): From 3924a90c7491ee719b94d867046526be09364dc7 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Sep 2015 16:43:28 +0200 Subject: [PATCH 05/36] Create conn_url propery in a `Leader` object --- helpers/dcs.py | 4 ++++ helpers/postgresql.py | 8 ++++---- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/helpers/dcs.py b/helpers/dcs.py index 22c47e26..f837245c 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -62,6 +62,10 @@ class Leader(namedtuple('Leader', 'index,expiration,ttl,member')): def name(self): return self.member.name + @property + def conn_url(self): + return self.member.conn_url + class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): """Immutable object (namedtuple) which represents PostgreSQL cluster. diff --git a/helpers/postgresql.py b/helpers/postgresql.py index ca4ea9c0..ce8ca18f 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -128,7 +128,7 @@ class Postgresql: os.path.exists(self.trigger_file) and os.unlink(self.trigger_file) def sync_from_leader(self, leader): - r = parseurl(leader.member.conn_url) + r = parseurl(leader.conn_url) pgpass = 'pgpass' with open(pgpass, 'w') as f: @@ -288,7 +288,7 @@ class Postgresql: if not os.path.isfile(self.recovery_conf): return False - pattern = leader and leader.member.conn_url and self.primary_conninfo(leader.member.conn_url) + pattern = leader and leader.conn_url and self.primary_conninfo(leader.conn_url) with open(self.recovery_conf, 'r') as f: for line in f: @@ -304,11 +304,11 @@ class Postgresql: f.write("""standby_mode = 'on' recovery_target_timeline = 'latest' """) - if leader and leader.member.conn_url: + if leader and leader.conn_url: f.write(""" primary_slot_name = '{}' primary_conninfo = '{}' -""".format(self.name, self.primary_conninfo(leader.member.conn_url))) +""".format(self.name, self.primary_conninfo(leader.conn_url))) for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) From eb85caa3bd230927803dfc4df8b0dcf1c21d4e4d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Sep 2015 17:16:15 +0200 Subject: [PATCH 06/36] Remove spurious print --- tests/test_etcd.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 6754a68a..c42394fa 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -63,7 +63,6 @@ def requests_get(url, **kwargs): def etcd_watch(key, index=None, timeout=None, recursive=None): - print ('watch', key, index, timeout) if timeout == 1: raise urllib3.exceptions.TimeoutError elif timeout == 5: From c6abf857400c6f9cc4ca51bdd4205e3d2ef76ac5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 2 Sep 2015 08:42:47 +0200 Subject: [PATCH 07/36] Bugfix: sleep method was renamed to watch --- patroni.py | 2 +- tests/test_patroni.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/patroni.py b/patroni.py index e26a4059..97d34ca2 100755 --- a/patroni.py +++ b/patroni.py @@ -76,7 +76,7 @@ class Patroni: if nap_time <= 0: self.next_run = current_time else: - self.ha.dcs.sleep(nap_time) + self.ha.dcs.watch(nap_time) def run(self): self.api.start() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index c0ea2907..489c4fa6 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -104,7 +104,7 @@ class TestPatroni(unittest.TestCase): self.p.touch_member = self.touch_member self.p.ha.state_handler.sync_replication_slots = time_sleep self.p.ha.dcs.client.read = etcd_read - self.p.ha.dcs.sleep = time_sleep + self.p.ha.dcs.watch = time_sleep self.assertRaises(SleepException, self.p.run) self.p.ha.state_handler.is_leader = lambda: False self.p.api.start = nop From 3e4d2df33bb4f78531da0a79496c89fe5f4ad83c Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Wed, 2 Sep 2015 12:04:39 +0200 Subject: [PATCH 08/36] put a real restore/replication command inside the test postgres.yaml files --- postgres0.yml | 2 +- postgres1.yml | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/postgres0.yml b/postgres0.yml index a2a7ce44..dd8f8356 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -46,7 +46,7 @@ postgresql: env_dir: /home/postgres/etc/wal-e.d/env threshold_megabytes: 10240 threshold_backup_size_percentage: 30 - restore: "true" + restore: scripts/restore.py #recovery_conf: #restore_command: cp ../wal_archive/%f %p parameters: diff --git a/postgres1.yml b/postgres1.yml index 6ef6b1c9..c2cb5ee6 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -48,6 +48,7 @@ postgresql: env_dir: /home/postgres/etc/wal-e.d/env threshold_megabytes: 10240 threshold_backup_size_percentage: 30 + restore: scripts/restore.py parameters: archive_mode: "on" wal_level: hot_standby From a5f2ab6f88bff36d0f056478a874467af938a830 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Wed, 2 Sep 2015 12:56:23 +0200 Subject: [PATCH 09/36] Better SIGCHLD handling, use correct user name. - Defer SIGCHLD processing until the later stage in order to allow subprocess to run its own reaping of children and return the correct exit code from the process called. - When changing the password at startup, do not assume the user postgres. --- helpers/postgresql.py | 2 +- helpers/utils.py | 37 +++++++++++++++++++++++-------------- patroni.py | 3 ++- 3 files changed, 26 insertions(+), 16 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index ce8ca18f..695e45c1 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -354,7 +354,7 @@ primary_conninfo = '{}' self.query('CREATE ROLE "{0}" WITH LOGIN SUPERUSER PASSWORD %s'.format( self.superuser['username']), self.superuser['password']) else: - self.query('ALTER ROLE postgres WITH PASSWORD %s', self.superuser['password']) + self.query('ALTER ROLE "{0}" WITH PASSWORD %s'.format(os.environ['USER']), self.superuser['password']) if self.admin: self.query('CREATE ROLE "{0}" WITH LOGIN CREATEDB CREATEROLE PASSWORD %s'.format( self.admin['username']), self.admin['password']) diff --git a/helpers/utils.py b/helpers/utils.py index 5f09490e..db578de5 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -5,7 +5,8 @@ import signal import sys import time -received_sigchld = False +interrupted_sleep = False +reap_children = False _DATE_TIME_RE = re.compile(r'''^ (?P\d{4})\-(?P\d{2})\-(?P\d{2}) # date @@ -72,30 +73,38 @@ def sigterm_handler(signo, stack_frame): def sigchld_handler(signo, stack_frame): - global received_sigchld - received_sigchld = True - try: - while True: - ret = os.waitpid(-1, os.WNOHANG) - if ret == (0, 0): - break - except OSError: - pass + global interrupted_sleep, reap_children + reap_children = interrupted_sleep = True def sleep(interval): - global received_sigchld + global interrupted_sleep current_time = time.time() end_time = current_time + interval while current_time < end_time: - received_sigchld = False + interrupted_sleep = False time.sleep(end_time - current_time) - if not received_sigchld: # we will ignore only sigchld + if not interrupted_sleep: # we will ignore only sigchld break current_time = time.time() - received_sigchld = False + interrupted_sleep = False def setup_signal_handlers(): signal.signal(signal.SIGTERM, sigterm_handler) signal.signal(signal.SIGCHLD, sigchld_handler) + + +def reap_children(): + global reap_children + if reap_children: + try: + while True: + ret = os.waitpid(-1, os.WNOHANG) + print ret + if ret == (0, 0): + break + except OSError: + pass + finally: + reap_children = False diff --git a/patroni.py b/patroni.py index e26a4059..a15f535b 100755 --- a/patroni.py +++ b/patroni.py @@ -9,7 +9,7 @@ from helpers.api import RestApiServer from helpers.etcd import Etcd from helpers.ha import Ha from helpers.postgresql import Postgresql -from helpers.utils import setup_signal_handlers, sleep +from helpers.utils import setup_signal_handlers, sleep, reap_children from helpers.zookeeper import ZooKeeper logger = logging.getLogger(__name__) @@ -93,6 +93,7 @@ class Patroni: except: logger.exception('Exception when changing replication slots') self.schedule_next_run() + reap_children() def main(): From 0beecb97a7d022b2b45a07188eb994b0f5b8ac32 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 2 Sep 2015 13:25:33 +0200 Subject: [PATCH 10/36] Reschedule next ha loop after promote --- patroni.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/patroni.py b/patroni.py index 97d34ca2..a711d5a8 100755 --- a/patroni.py +++ b/patroni.py @@ -70,6 +70,8 @@ class Patroni: self.postgresql.load_replication_slots() def schedule_next_run(self): + if self.postgresql.is_promoted: + self.next_run = time.time() self.next_run += self.nap_time current_time = time.time() nap_time = self.next_run - current_time From 8bc9d003911e1bdc9c6d3ef5d4590ec45682de29 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 2 Sep 2015 13:26:40 +0200 Subject: [PATCH 11/36] Code cleanup --- helpers/etcd.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index d6739e2d..67ac9c92 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -183,15 +183,13 @@ class Etcd(AbstractDCS): leader = Leader(leader.modifiedIndex, leader.expiration, leader.ttl, member) self.cluster = Cluster(initialize, leader, last_leader_operation, members) - return self.cluster except etcd.EtcdKeyNotFound: self.cluster = Cluster(False, None, None, []) - return self.cluster except: + self.cluster = None logger.exception('get_cluster') - - self.cluster = None - raise EtcdError('Etcd is not responding properly') + raise EtcdError('Etcd is not responding properly') + return self.cluster @catch_etcd_errors def touch_member(self, connection_string, ttl=None): From a7f7a7bf58231a3678ac7ee8c17473a800944040 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 2 Sep 2015 13:33:07 +0200 Subject: [PATCH 12/36] Remove spurious print --- helpers/utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/helpers/utils.py b/helpers/utils.py index db578de5..c725e6d4 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -101,7 +101,6 @@ def reap_children(): try: while True: ret = os.waitpid(-1, os.WNOHANG) - print ret if ret == (0, 0): break except OSError: From da132ca1fcc6b408652c923bf0fd353f67644cb4 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Wed, 2 Sep 2015 15:20:16 +0200 Subject: [PATCH 13/36] Do not rely on env vars for superuser role name. Environment variables may not be set in Docker. Instead, obtain the superuser role from the database. --- helpers/postgresql.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 695e45c1..8c9fd3c9 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -354,7 +354,8 @@ primary_conninfo = '{}' self.query('CREATE ROLE "{0}" WITH LOGIN SUPERUSER PASSWORD %s'.format( self.superuser['username']), self.superuser['password']) else: - self.query('ALTER ROLE "{0}" WITH PASSWORD %s'.format(os.environ['USER']), self.superuser['password']) + rolsuper = self.query("""SELECT rolname FROM pg_authid WHERE rolsuper = 't'""").fetchone()[0] + self.query('ALTER ROLE "{0}" WITH PASSWORD %s'.format(rolsuper), self.superuser['password']) if self.admin: self.query('CREATE ROLE "{0}" WITH LOGIN CREATEDB CREATEROLE PASSWORD %s'.format( self.admin['username']), self.admin['password']) From a670b598f4416b0e99acd7442e5a51295a5daaa3 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 2 Sep 2015 16:48:40 +0200 Subject: [PATCH 14/36] Implement unit test for reap_children function --- tests/test_utils.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/test_utils.py b/tests/test_utils.py index 76dbff66..312277b6 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,7 +2,7 @@ import os import time import unittest -from helpers.utils import sigchld_handler, sigterm_handler, sleep +from helpers.utils import reap_children, sigchld_handler, sigterm_handler, sleep def nop(*args, **kwargs): @@ -34,10 +34,11 @@ class TestUtils(unittest.TestCase): def test_sigterm_handler(self): self.assertRaises(SystemExit, sigterm_handler, None, None) - def test_sigchld_handler(self): - sigchld_handler(None, None) + def test_reap_children(self): + reap_children() os.waitpid = os_waitpid sigchld_handler(None, None) + reap_children() def test_sleep(self): time.sleep = time_sleep From 58410db9dd804597b6c45cb5ec329f09eb362154 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 11:47:08 +0200 Subject: [PATCH 15/36] Implemented Retry class inspired by KazooRetry --- helpers/__init__.py | 16 ++++++++++ helpers/dcs.py | 15 +-------- helpers/utils.py | 77 +++++++++++++++++++++++++++++++++++++++++++++ tests/test_utils.py | 58 +++++++++++++++++++++++++++++++++- 4 files changed, 151 insertions(+), 15 deletions(-) diff --git a/helpers/__init__.py b/helpers/__init__.py index e69de29b..2fe60065 100644 --- a/helpers/__init__.py +++ b/helpers/__init__.py @@ -0,0 +1,16 @@ +class PatroniException(Exception): + pass + + +class DCSError(PatroniException): + """Parent class for all kind of exceptions related to selected distributed configuration store""" + + def __init__(self, value): + self.value = value + + def __str__(self): + """ + >>> str(DCSError('foo')) + "'foo'" + """ + return repr(self.value) diff --git a/helpers/dcs.py b/helpers/dcs.py index f837245c..7ee97ca3 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -1,6 +1,7 @@ import abc from collections import namedtuple +from helpers import DCSError from helpers.utils import calculate_ttl, sleep from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl @@ -22,20 +23,6 @@ def parse_connection_string(value): return conn_url, api_url -class DCSError(Exception): - """Parent class for all kind of exceptions related to selected distributed configuration store""" - - def __init__(self, value): - self.value = value - - def __str__(self): - """ - >>> str(DCSError('foo')) - "'foo'" - """ - return repr(self.value) - - class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')): """Immutable object (namedtuple) which represents single member of PostgreSQL cluster. Consists of the following fields: diff --git a/helpers/utils.py b/helpers/utils.py index c725e6d4..f0555f48 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -1,10 +1,13 @@ import datetime import os +import random import re import signal import sys import time +from helpers import DCSError + interrupted_sleep = False reap_children = False @@ -107,3 +110,77 @@ def reap_children(): pass finally: reap_children = False + + +class RetryFailedError(DCSError): + """Raised when retrying an operation ultimately failed, after retrying the maximum number of attempts.""" + + +class Retry: + """Helper for retrying a method in the face of retry-able exceptions""" + + def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.8, max_delay=3600, + sleep_func=time.sleep, deadline=None, retry_exceptions=DCSError): + """Create a :class:`Retry` instance for retrying function calls + + :param max_tries: How many times to retry the command. -1 means infinite tries. + :param delay: Initial delay between retry attempts. + :param backoff: Backoff multiplier between retry attempts. Defaults to 2 for exponential backoff. + :param max_jitter: Additional max jitter period to wait between retry attempts to avoid slamming the server. + :param max_delay: Maximum delay in seconds, regardless of other backoff settings. Defaults to one hour. + :param retry_exceptions: single exception or tuple""" + + self.max_tries = max_tries + self.delay = delay + self.backoff = backoff + self.max_jitter = int(max_jitter * 100) + self.max_delay = float(max_delay) + self._attempts = 0 + self._cur_delay = delay + self.deadline = deadline + self._cur_stoptime = None + self.sleep_func = sleep_func + self.retry_exceptions = retry_exceptions + + def reset(self): + """Reset the attempt counter""" + self._attempts = 0 + self._cur_delay = self.delay + self._cur_stoptime = None + + def copy(self): + """Return a clone of this retry manager""" + return Retry(max_tries=self.max_tries, delay=self.delay, backoff=self.backoff, + max_jitter=self.max_jitter / 100.0, max_delay=self.max_delay, sleep_func=self.sleep_func, + deadline=self.deadline, retry_exceptions=self.retry_exceptions) + + def __call__(self, func, *args, **kwargs): + """Call a function with arguments until it completes without throwing a `retry_exceptions` + + :param func: Function to call + :param args: Positional arguments to call the function with + :params kwargs: Keyword arguments to call the function with + + The function will be called until it doesn't throw one of the retryable exceptions""" + self.reset() + + while True: + try: + if self.deadline is not None and self._cur_stoptime is None: + self._cur_stoptime = time.time() + self.deadline + return func(*args, **kwargs) + except self.retry_exceptions: + # Note: max_tries == -1 means infinite tries. + if self._attempts == self.max_tries: + raise RetryFailedError("Too many retry attempts") + self._attempts += 1 + sleeptime = self._cur_delay + ( + random.randint(0, self.max_jitter) / 100.0) + + if self._cur_stoptime is not None and \ + time.time() + sleeptime >= self._cur_stoptime: + raise RetryFailedError("Exceeded retry deadline") + else: + self.sleep_func(sleeptime) + self._cur_delay = min(self._cur_delay * self.backoff, + self.max_delay) diff --git a/tests/test_utils.py b/tests/test_utils.py index 312277b6..da43ada3 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,7 +2,8 @@ import os import time import unittest -from helpers.utils import reap_children, sigchld_handler, sigterm_handler, sleep +from helpers import DCSError +from helpers.utils import Retry, RetryFailedError, reap_children, sigchld_handler, sigterm_handler, sleep def nop(*args, **kwargs): @@ -43,3 +44,58 @@ class TestUtils(unittest.TestCase): def test_sleep(self): time.sleep = time_sleep sleep(0.01) + + +class TestRetrySleeper(unittest.TestCase): + + def _pass(self): + pass + + def _fail(self, times=1): + scope = dict(times=0) + + def inner(): + if scope['times'] >= times: + pass + else: + scope['times'] += 1 + raise DCSError('Failed!') + return inner + + def _makeOne(self, *args, **kwargs): + return Retry(*args, **kwargs) + + def test_reset(self): + retry = self._makeOne(delay=0, max_tries=2) + retry(self._fail()) + self.assertEquals(retry._attempts, 1) + retry.reset() + self.assertEquals(retry._attempts, 0) + + def test_too_many_tries(self): + retry = self._makeOne(delay=0) + self.assertRaises(RetryFailedError, retry, self._fail(times=999)) + self.assertEquals(retry._attempts, 1) + + def test_maximum_delay(self): + def sleep_func(_time): + pass + + retry = self._makeOne(delay=10, max_tries=100, sleep_func=sleep_func) + retry(self._fail(times=10)) + self.assertTrue(retry._cur_delay < 4000, retry._cur_delay) + # gevent's sleep function is picky about the type + self.assertEquals(type(retry._cur_delay), float) + + def test_deadline(self): + def sleep_func(_time): + pass + + retry = self._makeOne(deadline=0.0001, sleep_func=sleep_func) + self.assertRaises(RetryFailedError, retry, self._fail(times=10)) + + def test_copy(self): + _sleep = lambda t: None + retry = self._makeOne(sleep_func=_sleep) + rcopy = retry.copy() + self.assertTrue(rcopy.sleep_func is _sleep) From e8b2fb112b04d009030a3010e5decdbf2f03d9a9 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 11:50:58 +0200 Subject: [PATCH 16/36] It is better to reap childs before running relatively long sleep --- patroni.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni.py b/patroni.py index a15f535b..3be2aa57 100755 --- a/patroni.py +++ b/patroni.py @@ -92,8 +92,8 @@ class Patroni: self.ha.state_handler.drop_replication_slots() except: logger.exception('Exception when changing replication slots') - self.schedule_next_run() reap_children() + self.schedule_next_run() def main(): From 66286733b2df8ed187e09aaa2a0259b86a4067cb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 15:03:09 +0200 Subject: [PATCH 17/36] Enable retry functionality for etcd with default deadline=10 seconds --- helpers/etcd.py | 40 +++++++++++++++++++++++++++++----------- 1 file changed, 29 insertions(+), 11 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 67ac9c92..226c500a 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -11,7 +11,7 @@ import urllib3 from dns.exception import DNSException from dns import resolver from helpers.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string -from helpers.utils import sleep +from helpers.utils import Retry, RetryFailedError, sleep from requests.exceptions import RequestException logger = logging.getLogger(__name__) @@ -63,7 +63,12 @@ class Client(etcd.Client): # try to workarond bug in python-etcd: https://github.com/jplana/python-etcd/issues/81 def _result_from_response(self, response): - response.data.decode('utf-8') + try: + response.data.decode('utf-8') + except urllib3.exceptions.TimeoutError: + raise + except Exception as e: + raise etcd.EtcdException('Unable to decode server response: %s' % e) return super(Client, self)._result_from_response(response) def _get_machines_cache_from_srv(self, discovery_srv): @@ -131,7 +136,7 @@ def catch_etcd_errors(func): def wrapper(*args, **kwargs): try: return not func(*args, **kwargs) is None - except etcd.EtcdException: + except (RetryFailedError, etcd.EtcdException): return False return wrapper @@ -142,9 +147,17 @@ class Etcd(AbstractDCS): super(Etcd, self).__init__(name, config) self.ttl = config['ttl'] self.member_ttl = config.get('member_ttl', 3600) + self._retry = Retry(deadline=10, max_delay=1, max_tries=-1, + retry_exceptions=(etcd.EtcdConnectionFailed, + etcd.EtcdLeaderElectionInProgress, + etcd.EtcdWatcherCleared, + etcd.EtcdEventIndexCleared)) self.client = self.get_etcd_client(config) self.cluster = None + def retry(self, *args, **kwargs): + return self._retry.copy()(*args, **kwargs) + def get_etcd_client(self, config): client = None while not client: @@ -162,7 +175,7 @@ class Etcd(AbstractDCS): def get_cluster(self): try: - result = self.client.read(self.client_path(''), recursive=True) + result = self.retry(self.client.read, self.client_path(''), recursive=True) nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves} # get initialize flag @@ -193,17 +206,22 @@ class Etcd(AbstractDCS): @catch_etcd_errors def touch_member(self, connection_string, ttl=None): - return self.client.set(self.client_path('/members/' + self._name), connection_string, ttl or self.member_ttl) + return self.retry(self.client.set, self.client_path('/members/' + self._name), + connection_string, ttl or self.member_ttl) @catch_etcd_errors def take_leader(self): return self.client.set(self.client_path('/leader'), self._name, self.ttl) - @catch_etcd_errors def attempt_to_acquire_leader(self): - ret = self.client.write(self.client_path('/leader'), self._name, ttl=self.ttl, prevExist=False) - ret or logger.info('Could not take out TTL lock') - return ret + try: + return not self.retry(self.client.write, self.client_path('/leader'), + self._name, ttl=self.ttl, prevExist=False) is None + except etcd.EtcdAlreadyExist: + logger.info('Could not take out TTL lock') + except (RetryFailedError, etcd.EtcdException): + pass + return False @catch_etcd_errors def write_leader_optime(self, state_handler): @@ -211,13 +229,13 @@ class Etcd(AbstractDCS): @catch_etcd_errors def update_leader(self, state_handler): - ret = self.client.test_and_set(self.client_path('/leader'), self._name, self._name, self.ttl) + ret = self.retry(self.client.test_and_set, self.client_path('/leader'), self._name, self._name, self.ttl) ret and self.write_leader_optime(state_handler) return ret @catch_etcd_errors def race(self, path): - return self.client.write(self.client_path(path), self._name, prevExist=False) + return self.retry(self.client.write, self.client_path(path), self._name, prevExist=False) @catch_etcd_errors def delete_leader(self): From 2f6399de27d2d7e4eb90d20dbfc8d6c9b35c6669 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 16:35:58 +0200 Subject: [PATCH 18/36] More unit-tests for etcd --- tests/test_etcd.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index c42394fa..2f993f85 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -26,6 +26,10 @@ class MockResponse: @property def data(self): + if self.content == 'TimeoutError': + raise urllib3.exceptions.TimeoutError + if self.content == 'Exception': + raise Exception return self.content @property @@ -76,6 +80,8 @@ def etcd_watch(key, index=None, timeout=None, recursive=None): def etcd_write(key, value, **kwargs): + if key == '/service/exists/leader': + raise etcd.EtcdAlreadyExist if key == '/service/test/leader': if kwargs.get('prevValue', None) == 'foo' or not kwargs.get('prevExist', True): return True @@ -190,6 +196,15 @@ class TestClient(unittest.TestCase): self.assertEquals(self.client.get_srv_record('blabla'), []) self.assertEquals(self.client.get_srv_record('exception'), []) + def test__result_from_response(self): + response = MockResponse() + response.content = 'TimeoutError' + self.assertRaises(urllib3.exceptions.TimeoutError, self.client._result_from_response, response) + response.content = 'Exception' + self.assertRaises(etcd.EtcdException, self.client._result_from_response, response) + response.content = '{}' + self.assertRaises(etcd.EtcdException, self.client._result_from_response, response) + def test__get_machines_cache_from_srv(self): self.client.get_srv_record = lambda e: [('localhost', 2380)] self.client._get_machines_cache_from_srv('blabla') @@ -242,6 +257,12 @@ class TestEtcd(unittest.TestCase): def test_take_leader(self): self.assertFalse(self.etcd.take_leader()) + def testattempt_to_acquire_leader(self): + self.etcd._base_path = '/service/exists' + self.assertFalse(self.etcd.attempt_to_acquire_leader()) + self.etcd._base_path = '/service/failed' + self.assertFalse(self.etcd.attempt_to_acquire_leader()) + def test_update_leader(self): self.assertTrue(self.etcd.update_leader(MockPostgresql())) From 50420771ce3531248c779e52dbf61a2e4672f41c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 16:53:15 +0200 Subject: [PATCH 19/36] Retry when executing take_leader method. --- helpers/etcd.py | 2 +- tests/test_etcd.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 226c500a..74ad36f8 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -211,7 +211,7 @@ class Etcd(AbstractDCS): @catch_etcd_errors def take_leader(self): - return self.client.set(self.client_path('/leader'), self._name, self.ttl) + return self.retry(self.client.set, self.client_path('/leader'), self._name, self.ttl) def attempt_to_acquire_leader(self): try: diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 2f993f85..f5d65e5f 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -202,7 +202,7 @@ class TestClient(unittest.TestCase): self.assertRaises(urllib3.exceptions.TimeoutError, self.client._result_from_response, response) response.content = 'Exception' self.assertRaises(etcd.EtcdException, self.client._result_from_response, response) - response.content = '{}' + response.content = b'{}' self.assertRaises(etcd.EtcdException, self.client._result_from_response, response) def test__get_machines_cache_from_srv(self): From 330f9023eecd0f5153a749a8208ac5324f2943df Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 17:46:41 -0700 Subject: [PATCH 20/36] Interim commit to make replication slots optional for 9.3 users. --- helpers/postgresql.py | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index ce8ca18f..a8d749be 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -43,6 +43,7 @@ class Postgresql: def __init__(self, config): self.config = config self.name = config['name'] + self.scope = config['scope'] self.listen_addresses, self.port = config['listen'].split(':') self.data_dir = config['data_dir'] @@ -50,6 +51,7 @@ class Postgresql: self.superuser = config['superuser'] self.admin = config['admin'] self.callback = config.get('callbacks', {}) + self.use_slots = config['use_slots'] self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'), os.path.join(self.data_dir, 'postgresql.conf')) @@ -365,22 +367,24 @@ primary_conninfo = '{}' 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'") - self.members = [r[0] for r in cursor] + if self.use_slots: + cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'") + self.members = [r[0] for r in cursor] def sync_replication_slots(self, members): - # drop unused slots - for slot in set(self.members) - set(members): - self.query("""SELECT pg_drop_replication_slot(%s) - WHERE EXISTS(SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + if self.use_slots: + # drop unused slots + for slot in set(self.members) - set(members): + self.query("""SELECT pg_drop_replication_slot(%s) + WHERE EXISTS(SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) - # create new slots - for slot in set(members) - set(self.members): - self.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) - self.members = members + # create new slots + for slot in set(members) - set(self.members): + self.query("""SELECT pg_create_physical_replication_slot(%s) + WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) + self.members = members def create_replication_slots(self, cluster): self.sync_replication_slots([m.name for m in cluster.members if m.name != self.name]) From 2b801a3cccc5ebc634889366bd5565a7f3be3604 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 18:03:25 -0700 Subject: [PATCH 21/36] Next 9.3 compatibility commit. Removed pg_lsn, since it's not available in 9.3. --- helpers/postgresql.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index a8d749be..78a77174 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -258,7 +258,7 @@ class Postgresql: member_conn.autocommit = True member_cursor = member_conn.cursor() member_cursor.execute( - "SELECT pg_is_in_recovery(), %s - (pg_last_xlog_replay_location() - '0/0000000'::pg_lsn)", + "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')", (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() @@ -307,10 +307,9 @@ class Postgresql: recovery_target_timeline = 'latest' """) if leader and leader.conn_url: - f.write(""" -primary_slot_name = '{}' -primary_conninfo = '{}' -""".format(self.name, self.primary_conninfo(leader.conn_url))) + f.write("""primary_conninfo = '{}'\n""".format(self.primary_conninfo(leader.conn_url))) + if self.use_slots: + f.write("""primary_slot_name = '{}'\n""".format(self.name)) for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) @@ -363,8 +362,8 @@ primary_conninfo = '{}' def xlog_position(self): 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] + THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000') + ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] def load_replication_slots(self): if self.use_slots: From f2338e074c006942c73b0f3f6b280895c6f0b65f Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 18:16:44 -0700 Subject: [PATCH 22/36] Added documentation, sample config for optional replication slots. --- README.md | 5 +++-- postgres0.yml | 1 + postgres1.yml | 1 + 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 2582eb37..cfa141f5 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *connect_address*: ip address + port through which Postgres is accessible from other nodes and applications. * *data_dir*: file path to initialize and store Postgres data files * *maximum_lag_on_failover*: the maximum bytes a follower may lag before it is not eligible become leader + * *use_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3. * *pg_hba*: list of lines which should be added to pg_hba.conf * *- host all all 0.0.0.0/0 md5* * *replication* @@ -84,8 +85,8 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *admin*: * *username*: admin username, user will be created during initialization. It would have CREATEDB and CREATEROLE privileges * *password*: admin password, user will be created during initialization. - * *recovery_conf*: configuration settings written to recovery.conf when configuring follower - * *parameters*: list of configuration settings for Postgres + * *recovery_conf*: additional configuration settings written to recovery.conf when configuring follower + * *parameters*: list of configuration settings for Postgres. Many of these are required for replication to work. ## Replication choices diff --git a/postgres0.yml b/postgres0.yml index a2a7ce44..15dff82c 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -30,6 +30,7 @@ postgresql: connect_address: 127.0.0.1:5432 data_dir: data/postgresql0 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes + use_slots: True pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 diff --git a/postgres1.yml b/postgres1.yml index 6ef6b1c9..90731fe0 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -30,6 +30,7 @@ postgresql: connect_address: 127.0.0.1:5433 data_dir: data/postgresql1 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes + use_slots: True pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 From f5627a498e4e46f5480b38020f8a23194314285e Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 18:41:30 -0700 Subject: [PATCH 23/36] Fixed test_postgresql.py to include use_slots. --- README.md | 2 +- tests/test_postgresql.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index cfa141f5..345c12c3 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *connect_address*: ip address + port through which Postgres is accessible from other nodes and applications. * *data_dir*: file path to initialize and store Postgres data files * *maximum_lag_on_failover*: the maximum bytes a follower may lag before it is not eligible become leader - * *use_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3. + * *use_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3, and you should comment out max_replication_slots. * *pg_hba*: list of lines which should be added to pg_hba.conf * *- host all all 0.0.0.0/0 md5* * *replication* diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index c4a3d6df..882886df 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -114,6 +114,7 @@ class TestPostgresql(unittest.TestCase): 'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'], 'superuser': {'password': ''}, 'admin': {'username': 'admin', 'password': 'admin'}, + 'use_slots' : True, 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, From 5f4a9ffabb77499f1565a1fa8e45eb2532d033dc Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:04:42 -0700 Subject: [PATCH 24/36] Text spacing changes in an attempt to get flake8 to stop complaining. --- helpers/postgresql.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 78a77174..2a6af912 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -43,7 +43,6 @@ class Postgresql: def __init__(self, config): self.config = config self.name = config['name'] - self.scope = config['scope'] self.listen_addresses, self.port = config['listen'].split(':') self.data_dir = config['data_dir'] @@ -257,8 +256,8 @@ class Postgresql: member_conn = psycopg2.connect(**r) member_conn.autocommit = True member_cursor = member_conn.cursor() - member_cursor.execute( - "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')", + member_cursor.execute("""SELECT pg_is_in_recovery(), + %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')""", (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() @@ -362,8 +361,8 @@ recovery_target_timeline = 'latest' def xlog_position(self): return self.query("""SELECT CASE WHEN pg_is_in_recovery() - THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000') - ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] + THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000') + ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] def load_replication_slots(self): if self.use_slots: From bdb1454e3505031ac609a26c228fa179288613fb Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:10:30 -0700 Subject: [PATCH 25/36] Another commit because flake8 is a huge waste of time. --- helpers/postgresql.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 2a6af912..24f903e9 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -256,9 +256,9 @@ class Postgresql: member_conn = psycopg2.connect(**r) member_conn.autocommit = True member_cursor = member_conn.cursor() - member_cursor.execute("""SELECT pg_is_in_recovery(), - %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')""", - (self.xlog_position(), )) + member_cursor.execute("""SELECT pg_is_in_recovery(), + %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')""", + (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() member_conn.close() From 2b62adae210f487ddf667f6787a81679a9b55a8c Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:22:59 -0700 Subject: [PATCH 26/36] Fixed inherited bug in is_healthiest_node() with LSN position. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 24f903e9..62c7bb7e 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -257,7 +257,7 @@ class Postgresql: member_conn.autocommit = True member_cursor = member_conn.cursor() member_cursor.execute("""SELECT pg_is_in_recovery(), - %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')""", + %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/00000')""", (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() From 6df56fc6ccfe68259bd0d1556843e72d7e9b65e6 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:41:33 -0700 Subject: [PATCH 27/36] Another try at the is_healthiest_node bugfix. --- helpers/postgresql.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 62c7bb7e..bc6aed5b 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -382,7 +382,8 @@ recovery_target_timeline = 'latest' self.query("""SELECT pg_create_physical_replication_slot(%s) WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = %s)""", slot, slot) - self.members = members + + self.members = members def create_replication_slots(self, cluster): self.sync_replication_slots([m.name for m in cluster.members if m.name != self.name]) From aeea7196bd1eebac4411c414786cf2175bad61da Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:46:06 -0700 Subject: [PATCH 28/36] Fixing flake8 issue. --- helpers/postgresql.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index bc6aed5b..dbdecb8f 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -382,8 +382,8 @@ recovery_target_timeline = 'latest' self.query("""SELECT pg_create_physical_replication_slot(%s) WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = %s)""", slot, slot) - - self.members = members + + self.members = members def create_replication_slots(self, cluster): self.sync_replication_slots([m.name for m in cluster.members if m.name != self.name]) From 5612cd0280ec016831efa0e461f7825d1bac3400 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 22:20:27 -0700 Subject: [PATCH 29/36] Changed use_slots to be backwards compatible by using config.get() per PR feedback. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index dbdecb8f..03e4f4ef 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -50,7 +50,7 @@ class Postgresql: self.superuser = config['superuser'] self.admin = config['admin'] self.callback = config.get('callbacks', {}) - self.use_slots = config['use_slots'] + self.use_slots = config.get('use_slots',True) self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'), os.path.join(self.data_dir, 'postgresql.conf')) From 936173272260288023dade904324067dbcc7e311 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 08:29:16 +0200 Subject: [PATCH 30/36] Fix pep8 formatting and unit-tests --- helpers/postgresql.py | 20 ++++++++++---------- tests/test_postgresql.py | 3 +-- 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index ffe93be3..db4d3cd8 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -50,7 +50,7 @@ class Postgresql: self.superuser = config['superuser'] self.admin = config['admin'] self.callback = config.get('callbacks', {}) - self.use_slots = config.get('use_slots',True) + self.use_slots = config.get('use_slots', True) self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'), os.path.join(self.data_dir, 'postgresql.conf')) @@ -256,9 +256,9 @@ class Postgresql: member_conn = psycopg2.connect(**r) member_conn.autocommit = True member_cursor = member_conn.cursor() - member_cursor.execute("""SELECT pg_is_in_recovery(), - %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/00000')""", - (self.xlog_position(), )) + member_cursor.execute( + "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0')", + (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() member_conn.close() @@ -362,8 +362,8 @@ recovery_target_timeline = 'latest' def xlog_position(self): return self.query("""SELECT CASE WHEN pg_is_in_recovery() - THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000') - ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] + THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0') + ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/0') END""").fetchone()[0] def load_replication_slots(self): if self.use_slots: @@ -375,14 +375,14 @@ recovery_target_timeline = 'latest' # drop unused slots for slot in set(self.members) - set(members): self.query("""SELECT pg_drop_replication_slot(%s) - WHERE EXISTS(SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + WHERE EXISTS(SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) # create new slots for slot in set(members) - set(self.members): self.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) self.members = members diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 30278f87..ade68d3a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -111,7 +111,6 @@ class TestPostgresql(unittest.TestCase): 'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'], 'superuser': {'password': ''}, 'admin': {'username': 'admin', 'password': 'admin'}, - 'use_slots' : True, 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, @@ -120,7 +119,7 @@ class TestPostgresql(unittest.TestCase): 'on_restart': 'true', 'on_role_change': 'true', 'on_reload': 'true' }, - 'restore': '/usr/bin/true'}) + 'restore': 'true'}) psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) From 8b9e99090fc2cf2c8808fea711c9e8f9db9e588e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 10:41:59 +0200 Subject: [PATCH 31/36] Move lsn_to_bytes and bytes_to_lsn into Postgresql class and make their behavior version specific --- helpers/postgresql.py | 33 +++++++++++++++++++++++++++++---- helpers/utils.py | 26 -------------------------- tests/test_postgresql.py | 1 + 3 files changed, 30 insertions(+), 30 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index db4d3cd8..f84f87a0 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -4,14 +4,10 @@ import psycopg2 import shlex import shutil import subprocess -import six from helpers.utils import sleep from six.moves.urllib_parse import urlparse -if six.PY3: - long = int - logger = logging.getLogger(__name__) ACTION_ON_START = "on_start" @@ -39,6 +35,7 @@ def parseurl(url): class Postgresql: + _SERVER_VERSION = 90400 def __init__(self, config): self.config = config @@ -85,6 +82,7 @@ class Postgresql: r = parseurl('postgres://{}/postgres'.format(self.local_address)) self._connection = psycopg2.connect(**r) self._connection.autocommit = True + self._SERVER_VERSION = self._connection.server_version return self._connection def _cursor(self): @@ -394,3 +392,30 @@ recovery_target_timeline = 'latest' def last_operation(self): return str(self.xlog_position()) + + @staticmethod + def lsn_to_bytes(value): + """ + >>> Postgresql.lsn_to_bytes('1/66000060') + 6006243424 + >>> Postgresql.lsn_to_bytes('j/66000060') + 0 + """ + try: + multiplier = 0xFF000000 if Postgresql._SERVER_VERSION < 90300 else 0x100000000 + e = value.split('/') + if len(e) == 2 and len(e[0]) > 0 and len(e[1]) > 0: + return int(e[0], 16) * multiplier + int(e[1], 16) + except ValueError: + return 0 + + @staticmethod + def bytes_to_lsn(value): + """ + >>> Postgresql.bytes_to_lsn(6006243424) + '1/66000060' + """ + divider = 0xFF000000 if Postgresql._SERVER_VERSION < 90300 else 0x100000000 + segment = value / divider + offset = value % divider + return '%X/%X' % (segment, offset) diff --git a/helpers/utils.py b/helpers/utils.py index c725e6d4..c0553374 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -42,32 +42,6 @@ def calculate_ttl(expiration): return int((expiration - now).total_seconds()) -def lsn_to_bytes(value): - """ - >>> lsn_to_bytes('1/66000060') - 6006243424 - >>> lsn_to_bytes('j/66000060') - 0 - """ - try: - e = value.split('/') - if len(e) == 2 and len(e[0]) > 0 and len(e[1]) > 0: - return (int(e[0], 16) << 32) | int(e[1], 16) - except ValueError: - pass - return 0 - - -def bytes_to_lsn(value): - """ - >>> bytes_to_lsn(6006243424) - '1/66000060' - """ - id = value >> 32 - off = value & 0xffffffff - return '%x/%x' % (id, off) - - def sigterm_handler(signo, stack_frame): sys.exit() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index ade68d3a..f8187c9f 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -79,6 +79,7 @@ class MockConnect: def __init__(self): self.autocommit = False self.closed = 0 + self.server_version = 90400 def cursor(self): return MockCursor() From c913c8ad9270ee3ae100ac3912d06d2ace6d395b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 12:11:55 +0200 Subject: [PATCH 32/36] Calculate xlog bytes with using lsn_to_bytes method This method behaves differently depending on server version and will allow to use patroni with postgres older than 9.3 --- helpers/postgresql.py | 27 +++++++++++++++------------ tests/test_postgresql.py | 19 +++++++++++-------- 2 files changed, 26 insertions(+), 20 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index f84f87a0..642c48f4 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -82,7 +82,7 @@ class Postgresql: r = parseurl('postgres://{}/postgres'.format(self.local_address)) self._connection = psycopg2.connect(**r) self._connection.autocommit = True - self._SERVER_VERSION = self._connection.server_version + Postgresql._SERVER_VERSION = self._connection.server_version return self._connection def _cursor(self): @@ -254,9 +254,7 @@ class Postgresql: member_conn = psycopg2.connect(**r) member_conn.autocommit = True member_cursor = member_conn.cursor() - member_cursor.execute( - "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0')", - (self.xlog_position(), )) + member_cursor.execute("SELECT pg_is_in_recovery(), COALESCE(pg_last_xlog_replay_location(), '0/0')") row = member_cursor.fetchone() member_cursor.close() member_conn.close() @@ -264,7 +262,7 @@ class Postgresql: if not row[0]: logger.warning('Master (%s) is still alive', member.name) return False - if row[1] < 0: + if self.xlog_position() < self.lsn_to_bytes(row[1], member_conn.server_version): return False except psycopg2.Error: continue @@ -359,9 +357,10 @@ recovery_target_timeline = 'latest' self.admin['username']), self.admin['password']) def xlog_position(self): - return self.query("""SELECT CASE WHEN pg_is_in_recovery() - THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0') - ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/0') END""").fetchone()[0] + lsn = self.query("""SELECT CASE WHEN pg_is_in_recovery() + THEN pg_last_xlog_replay_location() + ELSE pg_current_xlog_location() END""").fetchone()[0] + return self.lsn_to_bytes(lsn) def load_replication_slots(self): if self.use_slots: @@ -394,15 +393,17 @@ recovery_target_timeline = 'latest' return str(self.xlog_position()) @staticmethod - def lsn_to_bytes(value): + def lsn_to_bytes(value, version=None): """ >>> Postgresql.lsn_to_bytes('1/66000060') 6006243424 >>> Postgresql.lsn_to_bytes('j/66000060') 0 """ + if version is None: + version = Postgresql._SERVER_VERSION try: - multiplier = 0xFF000000 if Postgresql._SERVER_VERSION < 90300 else 0x100000000 + multiplier = 0xFF000000 if version < 90300 else 0x100000000 e = value.split('/') if len(e) == 2 and len(e[0]) > 0 and len(e[1]) > 0: return int(e[0], 16) * multiplier + int(e[1], 16) @@ -410,12 +411,14 @@ recovery_target_timeline = 'latest' return 0 @staticmethod - def bytes_to_lsn(value): + def bytes_to_lsn(value, version=None): """ >>> Postgresql.bytes_to_lsn(6006243424) '1/66000060' """ - divider = 0xFF000000 if Postgresql._SERVER_VERSION < 90300 else 0x100000000 + if version is None: + version = Postgresql._SERVER_VERSION + divider = 0xFF000000 if version < 90300 else 0x100000000 segment = value / divider offset = value % divider return '%X/%X' % (segment, offset) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index f8187c9f..a63c240b 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -21,13 +21,14 @@ def false(*args, **kwargs): class MockCursor: + _count = 0 def __init__(self): self.closed = False - self.current = 0 self.results = [] def execute(self, sql, *params): + MockCursor._count += 1 if sql.startswith('blabla'): raise psycopg2.OperationalError() elif sql.startswith('InterfaceError'): @@ -36,15 +37,15 @@ class MockCursor: self.results = [('blabla',), ('foobar',)] elif sql.startswith('SELECT pg_current_xlog_location()'): self.results = [(0,)] - elif sql.startswith('SELECT pg_is_in_recovery(), %s'): - if params[0][0] == 1: + elif sql.startswith('SELECT pg_is_in_recovery(), COALESCE'): + if MockCursor._count == 1: raise psycopg2.OperationalError() - elif params[0][0] == 2: - self.results = [(True, -1)] + elif MockCursor._count == 2: + self.results = [(True, '0/1')] else: - self.results = [(False, 0)] + self.results = [(False, '0/1')] elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'): - self.results = [(0,)] + self.results = [('0/0', )] elif sql.startswith('SELECT pg_is_in_recovery()'): self.results = [(False, )] elif sql.startswith('SELECT to_char(pg_postmaster_start_time'): @@ -184,12 +185,14 @@ class TestPostgresql(unittest.TestCase): 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 + MockCursor._count = 0 self.assertFalse(self.p.is_healthiest_node(cluster)) + MockCursor._count = 0 self.p.xlog_position = lambda: 1 self.assertTrue(self.p.is_healthiest_node(cluster)) self.p.xlog_position = lambda: 2 self.assertFalse(self.p.is_healthiest_node(cluster)) - self.p.config['maximum_lag_on_failover'] = -2 + self.p.config['maximum_lag_on_failover'] = -3 self.assertFalse(self.p.is_healthiest_node(cluster)) def test_is_leader(self): From a69565fc5f6a002c97faea8e6e544ecb17aeae73 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 13:22:44 +0200 Subject: [PATCH 33/36] Set Postgresql._SERVER_VERSION only after real connect. --- helpers/postgresql.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 642c48f4..3fc7983a 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -82,7 +82,7 @@ class Postgresql: r = parseurl('postgres://{}/postgres'.format(self.local_address)) self._connection = psycopg2.connect(**r) self._connection.autocommit = True - Postgresql._SERVER_VERSION = self._connection.server_version + Postgresql._SERVER_VERSION = self._connection.server_version return self._connection def _cursor(self): @@ -358,8 +358,8 @@ recovery_target_timeline = 'latest' def xlog_position(self): lsn = self.query("""SELECT CASE WHEN pg_is_in_recovery() - THEN pg_last_xlog_replay_location() - ELSE pg_current_xlog_location() END""").fetchone()[0] + THEN pg_last_xlog_replay_location() + ELSE pg_current_xlog_location() END""").fetchone()[0] return self.lsn_to_bytes(lsn) def load_replication_slots(self): From 650e2449042763ccac6bd923f4549db9a5b31253 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 16:06:44 +0200 Subject: [PATCH 34/36] Refactor directory structure in preparation for building pypi-package --- patroni.py | 120 +---------------------- patroni/__init__.py | 119 ++++++++++++++++++++++ patroni/__main__.py | 5 + {helpers => patroni}/api.py | 0 {helpers => patroni}/dcs.py | 2 +- {helpers => patroni}/etcd.py | 4 +- {helpers => patroni}/ha.py | 2 +- {helpers => patroni}/postgresql.py | 2 +- {helpers => patroni/scripts}/__init__.py | 0 {scripts => patroni/scripts}/aws.py | 0 {scripts => patroni/scripts}/restore.py | 0 {helpers => patroni}/utils.py | 0 patroni/version.py | 1 + {helpers => patroni}/zookeeper.py | 4 +- scripts/__init__.py | 0 setup.py | 18 ++-- tests/test_api.py | 2 +- tests/test_aws.py | 2 +- tests/test_etcd.py | 4 +- tests/test_ha.py | 6 +- tests/test_patroni.py | 12 +-- tests/test_postgresql.py | 4 +- tests/test_restore.py | 2 +- tests/test_utils.py | 2 +- tests/test_zookeeper.py | 8 +- 25 files changed, 166 insertions(+), 153 deletions(-) create mode 100644 patroni/__init__.py create mode 100644 patroni/__main__.py rename {helpers => patroni}/api.py (100%) rename {helpers => patroni}/dcs.py (99%) rename {helpers => patroni}/etcd.py (98%) rename {helpers => patroni}/ha.py (99%) rename {helpers => patroni}/postgresql.py (99%) rename {helpers => patroni/scripts}/__init__.py (100%) rename {scripts => patroni/scripts}/aws.py (100%) rename {scripts => patroni/scripts}/restore.py (100%) rename {helpers => patroni}/utils.py (100%) create mode 100644 patroni/version.py rename {helpers => patroni}/zookeeper.py (98%) delete mode 100644 scripts/__init__.py diff --git a/patroni.py b/patroni.py index 3be2aa57..f34ef133 100755 --- a/patroni.py +++ b/patroni.py @@ -1,123 +1,5 @@ #!/usr/bin/env python -import logging -import os -import sys -import time -import yaml - -from helpers.api import RestApiServer -from helpers.etcd import Etcd -from helpers.ha import Ha -from helpers.postgresql import Postgresql -from helpers.utils import setup_signal_handlers, sleep, reap_children -from helpers.zookeeper import ZooKeeper - -logger = logging.getLogger(__name__) - - -class Patroni: - - def __init__(self, config): - self.nap_time = config['loop_wait'] - self.postgresql = Postgresql(config['postgresql']) - self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config)) - host, port = config['restapi']['listen'].split(':') - self.api = RestApiServer(self, config['restapi']) - self.next_run = time.time() - self.shutdown_member_ttl = 300 - - @staticmethod - def get_dcs(name, config): - if 'etcd' in config: - return Etcd(name, config['etcd']) - if 'zookeeper' in config: - return ZooKeeper(name, config['zookeeper']) - raise Exception('Can not find sutable configuration of distributed configuration store') - - 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.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: - return True - return self.ha.dcs.touch_member(connection_string, ttl) - - def initialize(self): - # wait for etcd to be available - while not self.touch_member(): - logger.info('waiting on DCS') - sleep(5) - - # is data directory empty? - if self.postgresql.data_directory_empty(): - # racing to initialize - if self.ha.dcs.race('/initialize'): - self.postgresql.initialize() - self.ha.dcs.take_leader() - self.postgresql.start() - self.postgresql.create_replication_user() - self.postgresql.create_connection_users() - else: - while True: - leader = self.ha.dcs.current_leader() - if leader and self.postgresql.sync_from_leader(leader): - self.postgresql.write_recovery_conf(leader) - self.postgresql.start() - break - sleep(5) - elif self.postgresql.is_running(): - self.postgresql.load_replication_slots() - - def schedule_next_run(self): - self.next_run += self.nap_time - current_time = time.time() - nap_time = self.next_run - current_time - if nap_time <= 0: - self.next_run = current_time - else: - self.ha.dcs.sleep(nap_time) - - def run(self): - self.api.start() - self.next_run = time.time() - - while True: - self.touch_member() - logger.info(self.ha.run_cycle()) - try: - if self.ha.state_handler.is_leader(): - self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster) - else: - self.ha.state_handler.drop_replication_slots() - except: - logger.exception('Exception when changing replication slots') - reap_children() - self.schedule_next_run() - - -def main(): - logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) - logging.getLogger('requests').setLevel(logging.WARNING) - setup_signal_handlers() - - if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]): - print('Usage: {} config.yml'.format(sys.argv[0])) - return - - with open(sys.argv[1], 'r') as f: - config = yaml.load(f) - - patroni = Patroni(config) - try: - patroni.initialize() - patroni.run() - except KeyboardInterrupt: - pass - finally: - patroni.touch_member(patroni.shutdown_member_ttl) # schedule member removal - patroni.postgresql.stop() - patroni.ha.dcs.delete_leader() +from patroni import main if __name__ == '__main__': diff --git a/patroni/__init__.py b/patroni/__init__.py new file mode 100644 index 00000000..572f4263 --- /dev/null +++ b/patroni/__init__.py @@ -0,0 +1,119 @@ +import logging +import os +import sys +import time +import yaml + +from patroni.api import RestApiServer +from patroni.etcd import Etcd +from patroni.ha import Ha +from patroni.postgresql import Postgresql +from patroni.utils import setup_signal_handlers, sleep, reap_children +from patroni.zookeeper import ZooKeeper + +logger = logging.getLogger(__name__) + + +class Patroni: + + def __init__(self, config): + self.nap_time = config['loop_wait'] + self.postgresql = Postgresql(config['postgresql']) + self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config)) + host, port = config['restapi']['listen'].split(':') + self.api = RestApiServer(self, config['restapi']) + self.next_run = time.time() + self.shutdown_member_ttl = 300 + + @staticmethod + def get_dcs(name, config): + if 'etcd' in config: + return Etcd(name, config['etcd']) + if 'zookeeper' in config: + return ZooKeeper(name, config['zookeeper']) + raise Exception('Can not find sutable configuration of distributed configuration store') + + 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.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: + return True + return self.ha.dcs.touch_member(connection_string, ttl) + + def initialize(self): + # wait for etcd to be available + while not self.touch_member(): + logger.info('waiting on DCS') + sleep(5) + + # is data directory empty? + if self.postgresql.data_directory_empty(): + # racing to initialize + if self.ha.dcs.race('/initialize'): + self.postgresql.initialize() + self.ha.dcs.take_leader() + self.postgresql.start() + self.postgresql.create_replication_user() + self.postgresql.create_connection_users() + else: + while True: + leader = self.ha.dcs.current_leader() + if leader and self.postgresql.sync_from_leader(leader): + self.postgresql.write_recovery_conf(leader) + self.postgresql.start() + break + sleep(5) + elif self.postgresql.is_running(): + self.postgresql.load_replication_slots() + + def schedule_next_run(self): + self.next_run += self.nap_time + current_time = time.time() + nap_time = self.next_run - current_time + if nap_time <= 0: + self.next_run = current_time + else: + self.ha.dcs.sleep(nap_time) + + def run(self): + self.api.start() + self.next_run = time.time() + + while True: + self.touch_member() + logger.info(self.ha.run_cycle()) + try: + if self.ha.state_handler.is_leader(): + self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster) + else: + self.ha.state_handler.drop_replication_slots() + except: + logger.exception('Exception when changing replication slots') + reap_children() + self.schedule_next_run() + + +def main(): + logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) + logging.getLogger('requests').setLevel(logging.WARNING) + setup_signal_handlers() + + if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]): + print('Usage: {} config.yml'.format(sys.argv[0])) + return + + with open(sys.argv[1], 'r') as f: + config = yaml.load(f) + + patroni = Patroni(config) + try: + patroni.initialize() + patroni.run() + except KeyboardInterrupt: + pass + finally: + patroni.touch_member(patroni.shutdown_member_ttl) # schedule member removal + patroni.postgresql.stop() + patroni.ha.dcs.delete_leader() diff --git a/patroni/__main__.py b/patroni/__main__.py new file mode 100644 index 00000000..3abcbfc3 --- /dev/null +++ b/patroni/__main__.py @@ -0,0 +1,5 @@ +from patroni import main + + +if __name__ == '__main__': + main() diff --git a/helpers/api.py b/patroni/api.py similarity index 100% rename from helpers/api.py rename to patroni/api.py diff --git a/helpers/dcs.py b/patroni/dcs.py similarity index 99% rename from helpers/dcs.py rename to patroni/dcs.py index c7140c22..f5eb4d2c 100644 --- a/helpers/dcs.py +++ b/patroni/dcs.py @@ -1,7 +1,7 @@ import abc from collections import namedtuple -from helpers.utils import calculate_ttl, sleep +from patroni.utils import calculate_ttl, sleep from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl diff --git a/helpers/etcd.py b/patroni/etcd.py similarity index 98% rename from helpers/etcd.py rename to patroni/etcd.py index add736c5..0ebb99f4 100644 --- a/helpers/etcd.py +++ b/patroni/etcd.py @@ -8,8 +8,8 @@ import socket from dns.exception import DNSException from dns import resolver -from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string -from helpers.utils import sleep +from patroni.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string +from patroni.utils import sleep from requests.exceptions import RequestException logger = logging.getLogger(__name__) diff --git a/helpers/ha.py b/patroni/ha.py similarity index 99% rename from helpers/ha.py rename to patroni/ha.py index 8e283c55..f31b2b26 100644 --- a/helpers/ha.py +++ b/patroni/ha.py @@ -1,6 +1,6 @@ import logging -from helpers.dcs import DCSError +from patroni.dcs import DCSError from psycopg2 import InterfaceError, OperationalError logger = logging.getLogger(__name__) diff --git a/helpers/postgresql.py b/patroni/postgresql.py similarity index 99% rename from helpers/postgresql.py rename to patroni/postgresql.py index 3fc7983a..2c86800d 100644 --- a/helpers/postgresql.py +++ b/patroni/postgresql.py @@ -5,7 +5,7 @@ import shlex import shutil import subprocess -from helpers.utils import sleep +from patroni.utils import sleep from six.moves.urllib_parse import urlparse logger = logging.getLogger(__name__) diff --git a/helpers/__init__.py b/patroni/scripts/__init__.py similarity index 100% rename from helpers/__init__.py rename to patroni/scripts/__init__.py diff --git a/scripts/aws.py b/patroni/scripts/aws.py similarity index 100% rename from scripts/aws.py rename to patroni/scripts/aws.py diff --git a/scripts/restore.py b/patroni/scripts/restore.py similarity index 100% rename from scripts/restore.py rename to patroni/scripts/restore.py diff --git a/helpers/utils.py b/patroni/utils.py similarity index 100% rename from helpers/utils.py rename to patroni/utils.py diff --git a/patroni/version.py b/patroni/version.py new file mode 100644 index 00000000..11d27f8c --- /dev/null +++ b/patroni/version.py @@ -0,0 +1 @@ +__version__ = '0.1' diff --git a/helpers/zookeeper.py b/patroni/zookeeper.py similarity index 98% rename from helpers/zookeeper.py rename to patroni/zookeeper.py index cb2918cd..bbc05002 100644 --- a/helpers/zookeeper.py +++ b/patroni/zookeeper.py @@ -3,10 +3,10 @@ import random import requests import time -from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string -from helpers.utils import sleep from kazoo.client import KazooClient, KazooState from kazoo.exceptions import NoNodeError, NodeExistsError +from patroni.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string +from patroni.utils import sleep from requests.exceptions import RequestException logger = logging.getLogger(__name__) diff --git a/scripts/__init__.py b/scripts/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/setup.py b/setup.py index 62d05a69..4ed8ca37 100644 --- a/setup.py +++ b/setup.py @@ -19,13 +19,20 @@ if sys.version_info < (2, 7, 0): __location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.currentframe()))) +def read_version(package): + data = {} + with open(os.path.join(package, 'version.py'), 'r') as fd: + exec(fd.read(), data) + return data['__version__'] + + NAME = 'patroni' -MAIN_PACKAGE = 'patroni.py' -HELPERS = 'helpers' +MAIN_PACKAGE = NAME SCRIPTS = 'scripts' +VERSION = read_version(MAIN_PACKAGE) VERSION = '0.1' DESCRIPTION = 'A Template for PostgreSQL HA with etcd' -LICENSE = 'The MIT License' +LICENSE = 'MIT License' COVERAGE_XML = True COVERAGE_HTML = False @@ -62,8 +69,7 @@ class PyTest(TestCommand): def finalize_options(self): TestCommand.finalize_options(self) if self.cov_xml or self.cov_html: - self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov', SCRIPTS, '--cov-report', - 'term-missing'] + self.cov = ['--cov', MAIN_PACKAGE, '--cov', MAIN_PACKAGE, '--cov-report', 'term-missing'] if self.cov_xml: self.cov.extend(['--cov-report', 'xml']) if self.cov_html: @@ -82,7 +88,7 @@ class PyTest(TestCommand): params['plugins'] = ['cov'] if self.junitxml: params['args'] += self.junitxml - params['args'] += ['--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s'] + params['args'] += ['--doctest-modules', MAIN_PACKAGE, '-s', '-vv'] errno = pytest.main(**params) sys.exit(errno) diff --git a/tests/test_api.py b/tests/test_api.py index 91b36943..aecf3df6 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,7 +1,7 @@ import psycopg2 import unittest -from helpers.api import RestApiHandler, RestApiServer +from patroni.api import RestApiHandler, RestApiServer from six import BytesIO as IO from test_postgresql import psycopg2_connect diff --git a/tests/test_aws.py b/tests/test_aws.py index 84d495fe..09c357f4 100644 --- a/tests/test_aws.py +++ b/tests/test_aws.py @@ -2,7 +2,7 @@ import unittest import requests import boto.ec2 from collections import namedtuple -from scripts.aws import AWSConnection +from patroni.scripts.aws import AWSConnection from requests.exceptions import RequestException diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 38692c52..bc88e6a7 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -8,9 +8,9 @@ import time import unittest from dns.exception import DNSException -from helpers.dcs import Cluster, DCSError, Member -from helpers.etcd import Client, Etcd from mock import Mock, patch +from patroni.dcs import Cluster, DCSError, Member +from patroni.etcd import Client, Etcd class MockResponse: diff --git a/tests/test_ha.py b/tests/test_ha.py index abfc4ac8..46aa9a0a 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1,9 +1,9 @@ import unittest -from helpers.dcs import Cluster, DCSError -from helpers.etcd import Client, Etcd -from helpers.ha import Ha from mock import Mock, patch +from patroni.dcs import Cluster, DCSError +from patroni.etcd import Client, Etcd +from patroni.ha import Ha from test_etcd import etcd_read, etcd_write diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 67ac1ffd..979e603c 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -1,5 +1,5 @@ import datetime -import helpers.zookeeper +import patroni.zookeeper import psycopg2 import subprocess import sys @@ -7,12 +7,12 @@ import time import unittest import yaml -from helpers.api import RestApiServer -from helpers.dcs import Cluster, Member -from helpers.etcd import Etcd -from helpers.zookeeper import ZooKeeper from mock import Mock, patch +from patroni.api import RestApiServer +from patroni.dcs import Cluster, Member +from patroni.etcd import Etcd from patroni import Patroni, main +from patroni.zookeeper import ZooKeeper from six.moves import BaseHTTPServer from test_etcd import Client, etcd_read, etcd_write from test_ha import true, false @@ -70,7 +70,7 @@ class TestPatroni(unittest.TestCase): Postgresql.write_recovery_conf = self.write_recovery_conf def test_get_dcs(self): - helpers.zookeeper.KazooClient = MockKazooClient + patroni.zookeeper.KazooClient = MockKazooClient self.assertIsInstance(self.p.get_dcs('', {'zookeeper': {'scope': '', 'hosts': ''}}), ZooKeeper) self.assertRaises(Exception, self.p.get_dcs, '', {}) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index a63c240b..23fd9d74 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -4,8 +4,8 @@ import shutil import subprocess import unittest -from helpers.dcs import Cluster, Member -from helpers.postgresql import Postgresql +from patroni.dcs import Cluster, Member +from patroni.postgresql import Postgresql def nop(*args, **kwargs): diff --git a/tests/test_restore.py b/tests/test_restore.py index 38b34c6a..2ffd8a58 100644 --- a/tests/test_restore.py +++ b/tests/test_restore.py @@ -1,7 +1,7 @@ import unittest from mock import MagicMock, patch import os -from scripts.restore import Restore, WALERestore +from patroni.scripts.restore import Restore, WALERestore def fake_cursor_fetchone(*args, **kwargs): diff --git a/tests/test_utils.py b/tests/test_utils.py index 312277b6..9a383ca0 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,7 +2,7 @@ import os import time import unittest -from helpers.utils import reap_children, sigchld_handler, sigterm_handler, sleep +from patroni.utils import reap_children, sigchld_handler, sigterm_handler, sleep def nop(*args, **kwargs): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 53f5ab82..bf06fd19 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -1,8 +1,8 @@ -import helpers.zookeeper +import patroni.zookeeper import requests import unittest -from helpers.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError +from patroni.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError from kazoo.client import KazooState from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.protocol.states import ZnodeStat @@ -105,7 +105,7 @@ class TestExhibitorEnsembleProvider(unittest.TestCase): def set_up(self): requests.get = requests_get - helpers.zookeeper.sleep = exhibitor_sleep + patroni.zookeeper.sleep = exhibitor_sleep def test_init(self): self.assertRaises(Exception, ExhibitorEnsembleProvider, ['localhost'], 8181) @@ -119,7 +119,7 @@ class TestZooKeeper(unittest.TestCase): def set_up(self): requests.get = requests_get - helpers.zookeeper.KazooClient = MockKazooClient + patroni.zookeeper.KazooClient = MockKazooClient self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'}) def test_session_listener(self): From 7cce02ae958a499aca707c87131db051710bb269 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sat, 5 Sep 2015 15:19:17 +0200 Subject: [PATCH 35/36] Always use pg_xlog_location_diff to calculate bytes written to xlog Drop unused functionality (lsn_to_bytes, bytes_to_lsn) Revert some changes. --- helpers/postgresql.py | 47 +++++++--------------------------------- tests/test_postgresql.py | 18 ++++++--------- 2 files changed, 15 insertions(+), 50 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 3fc7983a..e451d844 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -35,7 +35,6 @@ def parseurl(url): class Postgresql: - _SERVER_VERSION = 90400 def __init__(self, config): self.config = config @@ -82,7 +81,6 @@ class Postgresql: r = parseurl('postgres://{}/postgres'.format(self.local_address)) self._connection = psycopg2.connect(**r) self._connection.autocommit = True - Postgresql._SERVER_VERSION = self._connection.server_version return self._connection def _cursor(self): @@ -254,7 +252,9 @@ class Postgresql: member_conn = psycopg2.connect(**r) member_conn.autocommit = True member_cursor = member_conn.cursor() - member_cursor.execute("SELECT pg_is_in_recovery(), COALESCE(pg_last_xlog_replay_location(), '0/0')") + member_cursor.execute( + "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')", + (self.xlog_position(),)) row = member_cursor.fetchone() member_cursor.close() member_conn.close() @@ -262,7 +262,7 @@ class Postgresql: if not row[0]: logger.warning('Master (%s) is still alive', member.name) return False - if self.xlog_position() < self.lsn_to_bytes(row[1], member_conn.server_version): + if row[1] < 0: return False except psycopg2.Error: continue @@ -357,10 +357,10 @@ recovery_target_timeline = 'latest' self.admin['username']), self.admin['password']) def xlog_position(self): - lsn = self.query("""SELECT CASE WHEN pg_is_in_recovery() - THEN pg_last_xlog_replay_location() - ELSE pg_current_xlog_location() END""").fetchone()[0] - return self.lsn_to_bytes(lsn) + return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery() + THEN pg_last_xlog_replay_location() + ELSE pg_current_xlog_location() + END, '0/0')""").fetchone()[0] def load_replication_slots(self): if self.use_slots: @@ -391,34 +391,3 @@ recovery_target_timeline = 'latest' def last_operation(self): return str(self.xlog_position()) - - @staticmethod - def lsn_to_bytes(value, version=None): - """ - >>> Postgresql.lsn_to_bytes('1/66000060') - 6006243424 - >>> Postgresql.lsn_to_bytes('j/66000060') - 0 - """ - if version is None: - version = Postgresql._SERVER_VERSION - try: - multiplier = 0xFF000000 if version < 90300 else 0x100000000 - e = value.split('/') - if len(e) == 2 and len(e[0]) > 0 and len(e[1]) > 0: - return int(e[0], 16) * multiplier + int(e[1], 16) - except ValueError: - return 0 - - @staticmethod - def bytes_to_lsn(value, version=None): - """ - >>> Postgresql.bytes_to_lsn(6006243424) - '1/66000060' - """ - if version is None: - version = Postgresql._SERVER_VERSION - divider = 0xFF000000 if version < 90300 else 0x100000000 - segment = value / divider - offset = value % divider - return '%X/%X' % (segment, offset) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index a63c240b..ca33f3e5 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -21,14 +21,12 @@ def false(*args, **kwargs): class MockCursor: - _count = 0 def __init__(self): self.closed = False self.results = [] def execute(self, sql, *params): - MockCursor._count += 1 if sql.startswith('blabla'): raise psycopg2.OperationalError() elif sql.startswith('InterfaceError'): @@ -37,15 +35,15 @@ class MockCursor: self.results = [('blabla',), ('foobar',)] elif sql.startswith('SELECT pg_current_xlog_location()'): self.results = [(0,)] - elif sql.startswith('SELECT pg_is_in_recovery(), COALESCE'): - if MockCursor._count == 1: + elif sql.startswith('SELECT pg_is_in_recovery(), %s'): + if params[0][0] == 1: raise psycopg2.OperationalError() - elif MockCursor._count == 2: - self.results = [(True, '0/1')] + elif params[0][0] == 2: + self.results = [(True, -1)] else: - self.results = [(False, '0/1')] - elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'): - self.results = [('0/0', )] + self.results = [(False, 0)] + elif sql.startswith('SELECT pg_xlog_location_diff'): + self.results = [(0,)] elif sql.startswith('SELECT pg_is_in_recovery()'): self.results = [(False, )] elif sql.startswith('SELECT to_char(pg_postmaster_start_time'): @@ -185,9 +183,7 @@ class TestPostgresql(unittest.TestCase): 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 - MockCursor._count = 0 self.assertFalse(self.p.is_healthiest_node(cluster)) - MockCursor._count = 0 self.p.xlog_position = lambda: 1 self.assertTrue(self.p.is_healthiest_node(cluster)) self.p.xlog_position = lambda: 2 From ac2740eeb91813ebd9d470a79d79c165aa5b5078 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sat, 5 Sep 2015 15:22:38 +0200 Subject: [PATCH 36/36] Drop unused variable server_version --- tests/test_postgresql.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index ca33f3e5..b3c91f73 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -78,7 +78,6 @@ class MockConnect: def __init__(self): self.autocommit = False self.closed = 0 - self.server_version = 90400 def cursor(self): return MockCursor()