From 3b1efff53e0db0e05b8bb7a2a4443994631293a2 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 27 Aug 2015 10:53:22 +0200 Subject: [PATCH 01/26] 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/26] 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/26] 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/26] 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/26] 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 e96265e4d157d2490455b6892a6f3c0819f3e07f Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 1 Sep 2015 17:05:05 +0200 Subject: [PATCH 06/26] Make Patroni suitable for packaging. - Restructured the repository: Moved patroni into its own directory. - Changed readme into reStructuredText --- README.md | 129 ------------ README.rst | 220 +++++++++++++++++++++ patroni/__init__.py | 1 + patroni/__main__.py | 4 + {helpers => patroni/helpers}/__init__.py | 0 {helpers => patroni/helpers}/api.py | 0 {helpers => patroni/helpers}/dcs.py | 2 +- {helpers => patroni/helpers}/etcd.py | 4 +- {helpers => patroni/helpers}/ha.py | 2 +- {helpers => patroni/helpers}/postgresql.py | 2 +- {helpers => patroni/helpers}/utils.py | 0 {helpers => patroni/helpers}/zookeeper.py | 4 +- patroni.py => patroni/patroni.py | 16 +- {scripts => patroni/scripts}/__init__.py | 0 {scripts => patroni/scripts}/aws.py | 0 {scripts => patroni/scripts}/restore.py | 0 release.sh | 28 +++ requirements-py3.txt | 4 +- setup.py | 27 ++- tests/test_api.py | 2 +- tests/test_aws.py | 2 +- tests/test_etcd.py | 4 +- tests/test_ha.py | 6 +- tests/test_patroni.py | 14 +- tests/test_postgresql.py | 4 +- tests/test_restore.py | 2 +- tests/test_utils.py | 2 +- tests/test_zookeeper.py | 8 +- 28 files changed, 310 insertions(+), 177 deletions(-) delete mode 100644 README.md create mode 100644 README.rst create mode 100644 patroni/__init__.py create mode 100644 patroni/__main__.py rename {helpers => patroni/helpers}/__init__.py (100%) rename {helpers => patroni/helpers}/api.py (100%) rename {helpers => patroni/helpers}/dcs.py (99%) rename {helpers => patroni/helpers}/etcd.py (98%) rename {helpers => patroni/helpers}/ha.py (99%) rename {helpers => patroni/helpers}/postgresql.py (99%) rename {helpers => patroni/helpers}/utils.py (100%) rename {helpers => patroni/helpers}/zookeeper.py (98%) rename patroni.py => patroni/patroni.py (93%) rename {scripts => patroni/scripts}/__init__.py (100%) rename {scripts => patroni/scripts}/aws.py (100%) rename {scripts => patroni/scripts}/restore.py (100%) create mode 100755 release.sh diff --git a/README.md b/README.md deleted file mode 100644 index 2582eb37..00000000 --- a/README.md +++ /dev/null @@ -1,129 +0,0 @@ -[![Build Status](https://travis-ci.org/zalando/patroni.svg?branch=master)](https://travis-ci.org/zalando/patroni) -[![Coverage Status](https://coveralls.io/repos/zalando/patroni/badge.svg?branch=master)](https://coveralls.io/r/zalando/patroni?branch=master) -# Patroni: A Template for PostgreSQL HA with ZooKeeper or etcd - -Patroni was previously known as Governor. - -*There are many ways to run high availability with PostgreSQL; here we present a template for you to create your own custom fit high availability solution using python and distributed configuration store (like ZooKeeper or etcd) for maximum accessibility.* - -## Getting Started -To get started, do the following from different terminals: - -``` -> etcd --data-dir=data/etcd -> ./patroni.py postgres0.yml -> ./patroni.py postgres1.yml -``` - -From there, you will see a high-availability cluster start up. Test -different settings in the YAML files to see how behavior changes. Kill -some of the different components to see how the system behaves. - -Add more `postgres*.yml` files to create an even larger cluster. - -We provide a haproxy configuration, which will give your application a single endpoint for connecting to the cluster's leader. To configure, run: - -``` -> haproxy -f haproxy.cfg -``` - -``` -> psql --host 127.0.0.1 --port 5000 postgres -``` - -## How Patroni works - -For a diagram of the high availability decision loop, see the included a PDF: [postgres-ha.pdf](https://github.com/zalando/patroni/blob/master/postgres-ha.pdf) - -## YAML Configuration - -For an example file, see `postgres0.yml`. Below is an explanation of settings: - -* *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated. -* *loop_wait*: the number of seconds the loop will sleep - -* *restapi* - * *listen*: ip address + port that Patroni will listen to provide health-check information for haproxy. - * *connect_address*: ip address + port through which restapi is accessible. - -* *etcd* - * *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd - * *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated. - * *host*: the host:port for the etcd endpoint - -* *zookeeper* - * *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd - * *session_timeout*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated. - * *reconnects_timeout*: how long we should try to reconnect to ZooKeeper after connection loss. After this timeout we assume that we don't have lock anymore and will restart in read-only mode. - * *hosts*: list of ZooKeeper cluster members in format: [ 'host1:port1', 'host2:port2', 'etc...'] - * *exhibitor*: if you are running ZooKeeper cluster under Exhibitor supervisory the following section could be interesting for you - * *poll_interval*: how often list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor - * *port*: Exhibitor port - * *hosts*: initial list of Exhibitor (ZooKeeper) nodes in format: [ 'host1', 'host2', 'etc...' ]. This list would be updated automatically when Exhibitor (ZooKeeper) cluster topology changes. - -* *postgresql* - * *name*: the name of the Postgres host, must be unique for the cluster - * *listen*: ip address + port that Postgres listening. Must be accessible from other nodes in the cluster if using streaming replication. - * *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 - * *pg_hba*: list of lines which should be added to pg_hba.conf - * *- host all all 0.0.0.0/0 md5* - * *replication* - * *username*: replication username, user will be created during initialization - * *password*: replication password, user will be created during initialization - * *network*: network setting for replication in pg_hba.conf - * *callbacks* callback scripts to run on certain actions. Patroni will pass current action, role and cluster name. See scripts/aws.py as an example on how to write them. - * *on_start*: a script to run when the cluster starts - * *on_stop*: a script to run when the cluster stops - * *on_restart*: a script to run when the cluster restarts - * *on_reload*: a script to run when configuration reload is triggered - * *on_role_change*: a script to run when the cluster is being promoted or demoted - * *superuser* - * *password*: password for postgres user. It would be set during initialization - * *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 - -## Replication choices - -Patroni uses Postgres' streaming replication. By default, this replication is asynchronous. For more information, see the [Postgres documentation on streaming replication](http://www.postgresql.org/docs/current/static/warm-standby.html#STREAMING-REPLICATION). - -Patroni's asynchronous replication configuration allows for `maximum_lag_on_failover` settings. This setting ensures failover will not occur if a follower is more than a certain number of bytes behind the follower. This setting should be increased or decreased based on business requirements. - -When asynchronous replication is not best for your use-case, investigate how Postgres's [synchronous replication](http://www.postgresql.org/docs/current/static/warm-standby.html#SYNCHRONOUS-REPLICATION) works. Synchronous replication ensures consistency across a cluster by confirming that writes are written to a secondary before returning to the connecting client with a success. The cost of synchronous replication will be reduced throughput on writes. This throughput will be entirely based on network performance. In hosted datacenter environments (like AWS, Rackspace, or any network you do not control), synchrous replication increases the variability of write performance significantly. If followers become inaccessible from the leader, the leader will becomes effectively readonly. - -To enable a simple synchronous replication test, add the follow lines to the `parameters` section of your YAML configuration files. - -```YAML - synchronous_commit: "on" - synchronous_standby_names: "*" -``` - -When using synchronous replication, use at least a 3-Postgres data nodes to ensure write availability if one host fails. - -Choosing your replication schema is dependent on the many business decisions. Investigate both async and sync replication, as well as other HA solutions, to determine which solution is best for you. - -## Applications should not use superusers - -When connecting from an application, always use a non-superuser. Patroni requires access to the database to function properly. By using a superuser from application, you can potentially use the entire connection pool, including the connections reserved for superusers with the `superuser_reserved_connections` setting. If Patroni cannot access the Primary, because the connection pool is full, behavior will be undesireable. - -## Requirements on a Mac - -Run the following on a Mac to install requirements: - -``` -brew install postgresql etcd haproxy libyaml python -pip install psycopg2 pyyaml -``` - -## Notice - -There are many different ways to do HA with PostgreSQL, see [the -PostgreSQL documentation](https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling) for a complete list. - -We call this project a "template" because it is far from a one-size fits -all, or a plug-and-play replication system. It will have it's own -caveats. Use wisely. diff --git a/README.rst b/README.rst new file mode 100644 index 00000000..c8c5eb7c --- /dev/null +++ b/README.rst @@ -0,0 +1,220 @@ +|Build Status| |Coverage Status| # Patroni: A Template for PostgreSQL HA +with ZooKeeper or etcd + +Patroni was previously known as Governor. + +*There are many ways to run high availability with PostgreSQL; here we +present a template for you to create your own custom fit high +availability solution using python and distributed configuration store +(like ZooKeeper or etcd) for maximum accessibility.* + +Getting Started +--------------- + +To get started, do the following from different terminals: + +:: + + > etcd --data-dir=data/etcd + > ./patroni.py postgres0.yml + > ./patroni.py postgres1.yml + +From there, you will see a high-availability cluster start up. Test +different settings in the YAML files to see how behavior changes. Kill +some of the different components to see how the system behaves. + +Add more ``postgres*.yml`` files to create an even larger cluster. + +We provide a haproxy configuration, which will give your application a +single endpoint for connecting to the cluster's leader. To configure, +run: + +:: + + > haproxy -f haproxy.cfg + +:: + + > psql --host 127.0.0.1 --port 5000 postgres + +How Patroni works +----------------- + +For a diagram of the high availability decision loop, see the included a +PDF: +`postgres-ha.pdf `__ + +YAML Configuration +------------------ + +For an example file, see ``postgres0.yml``. Below is an explanation of +settings: + +- *ttl*: the TTL to acquire the leader lock. Think of it as the length + of time before automatic failover process is initiated. +- *loop\_wait*: the number of seconds the loop will sleep + +- *restapi* +- *listen*: ip address + port that Patroni will listen to provide + health-check information for haproxy. +- *connect\_address*: ip address + port through which restapi is + accessible. + +- *etcd* +- *scope*: the relative path used on etcd's http api for this + deployment, thus you can run multiple HA deployments from a single + etcd +- *ttl*: the TTL to acquire the leader lock. Think of it as the length + of time before automatic failover process is initiated. +- *host*: the host:port for the etcd endpoint + +- *zookeeper* +- *scope*: the relative path used on etcd's http api for this + deployment, thus you can run multiple HA deployments from a single + etcd +- *session\_timeout*: the TTL to acquire the leader lock. Think of it + as the length of time before automatic failover process is initiated. +- *reconnects\_timeout*: how long we should try to reconnect to + ZooKeeper after connection loss. After this timeout we assume that we + don't have lock anymore and will restart in read-only mode. +- *hosts*: list of ZooKeeper cluster members in format: [ + 'host1:port1', 'host2:port2', 'etc...'] +- *exhibitor*: if you are running ZooKeeper cluster under Exhibitor + supervisory the following section could be interesting for you + + - *poll\_interval*: how often list of ZooKeeper and Exhibitor nodes + should be updated from Exhibitor + - *port*: Exhibitor port + - *hosts*: initial list of Exhibitor (ZooKeeper) nodes in format: [ + 'host1', 'host2', 'etc...' ]. This list would be updated + automatically when Exhibitor (ZooKeeper) cluster topology changes. + +- *postgresql* +- *name*: the name of the Postgres host, must be unique for the cluster +- *listen*: ip address + port that Postgres listening. Must be + accessible from other nodes in the cluster if using streaming + replication. +- *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 +- *pg\_hba*: list of lines which should be added to pg\_hba.conf + + - *- host all all 0.0.0.0/0 md5* + +- *replication* + + - *username*: replication username, user will be created during + initialization + - *password*: replication password, user will be created during + initialization + - *network*: network setting for replication in pg\_hba.conf + +- *callbacks* callback scripts to run on certain actions. Patroni will + pass current action, role and cluster name. See scripts/aws.py as an + example on how to write them. + + - *on\_start*: a script to run when the cluster starts + - *on\_stop*: a script to run when the cluster stops + - *on\_restart*: a script to run when the cluster restarts + - *on\_reload*: a script to run when configuration reload is + triggered + - *on\_role\_change*: a script to run when the cluster is being + promoted or demoted + +- *superuser* + + - *password*: password for postgres user. It would be set during + initialization + +- *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 + +Replication choices +------------------- + +Patroni uses Postgres' streaming replication. By default, this +replication is asynchronous. For more information, see the `Postgres +documentation on streaming +replication `__. + +Patroni's asynchronous replication configuration allows for +``maximum_lag_on_failover`` settings. This setting ensures failover will +not occur if a follower is more than a certain number of bytes behind +the follower. This setting should be increased or decreased based on +business requirements. + +When asynchronous replication is not best for your use-case, investigate +how Postgres's `synchronous +replication `__ +works. Synchronous replication ensures consistency across a cluster by +confirming that writes are written to a secondary before returning to +the connecting client with a success. The cost of synchronous +replication will be reduced throughput on writes. This throughput will +be entirely based on network performance. In hosted datacenter +environments (like AWS, Rackspace, or any network you do not control), +synchrous replication increases the variability of write performance +significantly. If followers become inaccessible from the leader, the +leader will becomes effectively readonly. + +To enable a simple synchronous replication test, add the follow lines to +the ``parameters`` section of your YAML configuration files. + +.. code:: YAML + + synchronous_commit: "on" + synchronous_standby_names: "*" + +When using synchronous replication, use at least a 3-Postgres data nodes +to ensure write availability if one host fails. + +Choosing your replication schema is dependent on the many business +decisions. Investigate both async and sync replication, as well as other +HA solutions, to determine which solution is best for you. + +Applications should not use superusers +-------------------------------------- + +When connecting from an application, always use a non-superuser. Patroni +requires access to the database to function properly. By using a +superuser from application, you can potentially use the entire +connection pool, including the connections reserved for superusers with +the ``superuser_reserved_connections`` setting. If Patroni cannot access +the Primary, because the connection pool is full, behavior will be +undesireable. + +Requirements on a Mac +--------------------- + +Run the following on a Mac to install requirements: + +:: + + brew install postgresql etcd haproxy libyaml python + pip install psycopg2 pyyaml + +Notice +------ + +There are many different ways to do HA with PostgreSQL, see `the +PostgreSQL +documentation `__ +for a complete list. + +We call this project a "template" because it is far from a one-size fits +all, or a plug-and-play replication system. It will have it's own +caveats. Use wisely. + +.. |Build Status| image:: https://travis-ci.org/zalando/patroni.svg?branch=master + :target: https://travis-ci.org/zalando/patroni +.. |Coverage Status| image:: https://coveralls.io/repos/zalando/patroni/badge.svg?branch=master + :target: https://coveralls.io/r/zalando/patroni?branch=master diff --git a/patroni/__init__.py b/patroni/__init__.py new file mode 100644 index 00000000..a3e6290d --- /dev/null +++ b/patroni/__init__.py @@ -0,0 +1 @@ +__version__ = '0.15' diff --git a/patroni/__main__.py b/patroni/__main__.py new file mode 100644 index 00000000..7390c1f1 --- /dev/null +++ b/patroni/__main__.py @@ -0,0 +1,4 @@ +import patroni + +if __name__ == '__main__': + patroni.main() diff --git a/helpers/__init__.py b/patroni/helpers/__init__.py similarity index 100% rename from helpers/__init__.py rename to patroni/helpers/__init__.py diff --git a/helpers/api.py b/patroni/helpers/api.py similarity index 100% rename from helpers/api.py rename to patroni/helpers/api.py diff --git a/helpers/dcs.py b/patroni/helpers/dcs.py similarity index 99% rename from helpers/dcs.py rename to patroni/helpers/dcs.py index c7140c22..69ccdebd 100644 --- a/helpers/dcs.py +++ b/patroni/helpers/dcs.py @@ -1,7 +1,7 @@ import abc from collections import namedtuple -from helpers.utils import calculate_ttl, sleep +from patroni.helpers.utils import calculate_ttl, sleep from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl diff --git a/helpers/etcd.py b/patroni/helpers/etcd.py similarity index 98% rename from helpers/etcd.py rename to patroni/helpers/etcd.py index add736c5..c58c1a78 100644 --- a/helpers/etcd.py +++ b/patroni/helpers/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.helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string +from patroni.helpers.utils import sleep from requests.exceptions import RequestException logger = logging.getLogger(__name__) diff --git a/helpers/ha.py b/patroni/helpers/ha.py similarity index 99% rename from helpers/ha.py rename to patroni/helpers/ha.py index 8e283c55..0496a094 100644 --- a/helpers/ha.py +++ b/patroni/helpers/ha.py @@ -1,6 +1,6 @@ import logging -from helpers.dcs import DCSError +from patroni.helpers.dcs import DCSError from psycopg2 import InterfaceError, OperationalError logger = logging.getLogger(__name__) diff --git a/helpers/postgresql.py b/patroni/helpers/postgresql.py similarity index 99% rename from helpers/postgresql.py rename to patroni/helpers/postgresql.py index ce8ca18f..8593fb52 100644 --- a/helpers/postgresql.py +++ b/patroni/helpers/postgresql.py @@ -6,7 +6,7 @@ import shutil import subprocess import six -from helpers.utils import sleep +from patroni.helpers.utils import sleep from six.moves.urllib_parse import urlparse if six.PY3: diff --git a/helpers/utils.py b/patroni/helpers/utils.py similarity index 100% rename from helpers/utils.py rename to patroni/helpers/utils.py diff --git a/helpers/zookeeper.py b/patroni/helpers/zookeeper.py similarity index 98% rename from helpers/zookeeper.py rename to patroni/helpers/zookeeper.py index cb2918cd..fd5f4f5b 100644 --- a/helpers/zookeeper.py +++ b/patroni/helpers/zookeeper.py @@ -3,8 +3,8 @@ import random import requests import time -from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string -from helpers.utils import sleep +from patroni.helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string +from patroni.helpers.utils import sleep from kazoo.client import KazooClient, KazooState from kazoo.exceptions import NoNodeError, NodeExistsError from requests.exceptions import RequestException diff --git a/patroni.py b/patroni/patroni.py similarity index 93% rename from patroni.py rename to patroni/patroni.py index e26a4059..dd7bd52e 100755 --- a/patroni.py +++ b/patroni/patroni.py @@ -5,12 +5,12 @@ 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 -from helpers.zookeeper import ZooKeeper +from patroni.helpers.api import RestApiServer +from patroni.helpers.etcd import Etcd +from patroni.helpers.ha import Ha +from patroni.helpers.postgresql import Postgresql +from patroni.helpers.utils import setup_signal_handlers, sleep +from patroni.helpers.zookeeper import ZooKeeper logger = logging.getLogger(__name__) @@ -117,7 +117,3 @@ def main(): patroni.touch_member(patroni.shutdown_member_ttl) # schedule member removal patroni.postgresql.stop() patroni.ha.dcs.delete_leader() - - -if __name__ == '__main__': - main() diff --git a/scripts/__init__.py b/patroni/scripts/__init__.py similarity index 100% rename from scripts/__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/release.sh b/release.sh new file mode 100755 index 00000000..5f97ba08 --- /dev/null +++ b/release.sh @@ -0,0 +1,28 @@ +#!/bin/sh + +if [ $# -ne 1 ]; then + >&2 echo "usage: $0 " + exit 1 +fi + +set -xe + +python3 --version +git --version + +version=$1 + +sed -i "s/__version__ = .*/__version__ = '${version}'/" __init__.py +python3 setup.py clean +python3 setup.py test +python3 setup.py flake8 + +git add __init__.py + +git commit -m "Bumped version to $version" +git push + +python3 setup.py sdist bdist_wheel upload + +git tag ${version} +git push --tags diff --git a/requirements-py3.txt b/requirements-py3.txt index 0fd9dfb3..544577ad 100644 --- a/requirements-py3.txt +++ b/requirements-py3.txt @@ -5,5 +5,5 @@ psycopg2 PyYAML requests six -kazoo>=2.2.1 -python-etcd>=0.4.1 +kazoo +python-etcd diff --git a/setup.py b/setup.py index 62d05a69..82ac8de8 100644 --- a/setup.py +++ b/setup.py @@ -18,14 +18,23 @@ 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, '__init__.py'), 'r') as fd: + exec(fd.read(), data) + return data['__version__'] NAME = 'patroni' -MAIN_PACKAGE = 'patroni.py' +MAIN_PACKAGE = 'patroni' HELPERS = 'helpers' SCRIPTS = 'scripts' -VERSION = '0.1' -DESCRIPTION = 'A Template for PostgreSQL HA with etcd' +VERSION = read_version(MAIN_PACKAGE) +DESCRIPTION = 'PostgreSQL High-Available orchestrator and CLI' LICENSE = 'The MIT License' +URL = 'https://github.com/zalando/patroni' +AUTHOR = 'Alexander Kukushkin, Alexey Klyukin, Feike Steenbergen' +AUTHOR_EMAIL = 'alexander.kukushkin@zalando.de, oleksii.kliukin@zalando.de, feike.steenbergen@zalando.de' +KEYWORDS = 'etcd governor patroni postgresql postgres ha zookeeper streaming replication' COVERAGE_XML = True COVERAGE_HTML = False @@ -38,7 +47,7 @@ CLASSIFIERS = [ 'Environment :: Console', 'Intended Audience :: Developers', 'Intended Audience :: System Administrators', - 'License :: OSI Approved :: The MIT License', + 'License :: OSI Approved :: MIT License', 'Operating System :: POSIX :: Linux', 'Programming Language :: Python', 'Programming Language :: Python :: 2.7', @@ -82,7 +91,8 @@ 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, '--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s'] + params['args'] += ['--doctest-modules', MAIN_PACKAGE, '-s', '-vv'] errno = pytest.main(**params) sys.exit(errno) @@ -118,10 +128,13 @@ def setup_package(): setup( name=NAME, version=version, + url=URL, + author=AUTHOR, + author_email=AUTHOR_EMAIL, description=DESCRIPTION, license=LICENSE, - keywords='etcd governor patroni postgresql postgres ha zookeeper', - long_description=read('README.md'), + keywords=KEYWORDS, + long_description=read('README.rst'), classifiers=CLASSIFIERS, test_suite='tests', packages=setuptools.find_packages(exclude=['tests', 'tests.*']), diff --git a/tests/test_api.py b/tests/test_api.py index 91b36943..12d7798b 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.helpers.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..3f09caa5 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -8,8 +8,8 @@ import time import unittest from dns.exception import DNSException -from helpers.dcs import Cluster, DCSError, Member -from helpers.etcd import Client, Etcd +from patroni.helpers.dcs import Cluster, DCSError, Member +from patroni.helpers.etcd import Client, Etcd from mock import Mock, patch diff --git a/tests/test_ha.py b/tests/test_ha.py index abfc4ac8..8f9a9c49 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1,8 +1,8 @@ import unittest -from helpers.dcs import Cluster, DCSError -from helpers.etcd import Client, Etcd -from helpers.ha import Ha +from patroni.helpers.dcs import Cluster, DCSError +from patroni.helpers.etcd import Client, Etcd +from patroni.helpers.ha import Ha from mock import Mock, patch from test_etcd import etcd_read, etcd_write diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 67ac1ffd..fd2e605d 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -1,5 +1,5 @@ import datetime -import helpers.zookeeper +import patroni.helpers.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 patroni.helpers.api import RestApiServer +from patroni.helpers.dcs import Cluster, Member +from patroni.helpers.etcd import Etcd +from patroni.helpers.zookeeper import ZooKeeper from mock import Mock, patch -from patroni import Patroni, main +from patroni.patroni import Patroni, main 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.helpers.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 040d1c66..f79a1438 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.helpers.dcs import Cluster, Member +from patroni.helpers.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 76dbff66..3bb238b0 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 patroni.helpers.utils import sigchld_handler, sigterm_handler, sleep def nop(*args, **kwargs): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 53f5ab82..1333897a 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -1,8 +1,8 @@ -import helpers.zookeeper +import patroni.helpers.zookeeper import requests import unittest -from helpers.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError +from patroni.helpers.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.helpers.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.helpers.zookeeper.KazooClient = MockKazooClient self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'}) def test_session_listener(self): From eb85caa3bd230927803dfc4df8b0dcf1c21d4e4d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Sep 2015 17:16:15 +0200 Subject: [PATCH 07/26] 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 08/26] 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 0beecb97a7d022b2b45a07188eb994b0f5b8ac32 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 2 Sep 2015 13:25:33 +0200 Subject: [PATCH 09/26] 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 10/26] 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 b1afd5ddc4b2e756846942349b8312e247348588 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Thu, 3 Sep 2015 09:52:01 +0200 Subject: [PATCH 11/26] Refactoring to enable package building and succesfull installation of the Patroni package. Build the first packages of patroni and added them to pypi.python.org Update Dockerfile to use pip to install patroni. --- MANIFEST.in | 3 +++ Dockerfile => docker/Dockerfile | 19 ++++++++----------- docker/Dockerfile.test.patch | 13 +++++++++++++ docker/entrypoint.sh | 2 +- patroni/__init__.py | 2 +- patroni/helpers/__init__.py | 0 patroni/patroni.py | 12 ++++++------ patroni/scripts/__init__.py | 0 requirements-py3.txt | 4 ++-- setup.py | 5 ++++- 10 files changed, 38 insertions(+), 22 deletions(-) create mode 100644 MANIFEST.in rename Dockerfile => docker/Dockerfile (71%) create mode 100644 docker/Dockerfile.test.patch delete mode 100644 patroni/helpers/__init__.py delete mode 100644 patroni/scripts/__init__.py diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 00000000..1a5e91cd --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,3 @@ +include requirements* +include *.rst +recursive-include patroni *.py diff --git a/Dockerfile b/docker/Dockerfile similarity index 71% rename from Dockerfile rename to docker/Dockerfile index 2068d856..1a1cb41c 100644 --- a/Dockerfile +++ b/docker/Dockerfile @@ -8,29 +8,26 @@ RUN apt-get update -y && apt-get install curl -y # Add PGDG repositories RUN echo "deb http://apt.postgresql.org/pub/repos/apt/ $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list -RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | sudo apt-key add - +RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add - RUN apt-get update -y RUN apt-get upgrade -y ENV PGVERSION 9.4 -RUN apt-get install python python-psycopg2 python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo python-pip -y -RUN pip install python-etcd +RUN apt-get install postgresql-${PGVERSION} -y +RUN apt-get install python python-psycopg2 python-yaml python-requests python-dnspython python-pip python-mock -y +RUN pip install patroni + ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH -RUN mkdir -p /patroni/helpers -RUN mkdir -p /patroni/scripts -ADD patroni.py /patroni/patroni.py -ADD helpers /patroni/helpers -ADD scripts /patroni/scripts - ENV ETCDVERSION 2.0.13 RUN curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz | tar xz -C /bin --strip=1 --wildcards --no-anchored etcd etcdctl -## Setting up a simple script that will serve as an entrypoint +### Setting up a simple script that will serve as an entrypoint +RUN mkdir /patroni/ RUN mkdir /data/ && touch /var/log/etcd.log /var/log/etcd.err /pgpass /patroni/postgres.yml RUN chown postgres:postgres -R /patroni/ /data/ /pgpass /var/log/etcd.* /patroni/postgres.yml -ADD docker/entrypoint.sh /entrypoint.sh +ADD entrypoint.sh /entrypoint.sh EXPOSE 4001 5432 2380 diff --git a/docker/Dockerfile.test.patch b/docker/Dockerfile.test.patch new file mode 100644 index 00000000..376f3c20 --- /dev/null +++ b/docker/Dockerfile.test.patch @@ -0,0 +1,13 @@ +--- Dockerfile 2015-09-03 09:24:16.355412216 +0200 ++++ Dockerfile.test 2015-09-03 09:25:48.407879693 +0200 +@@ -15,7 +15,9 @@ + ENV PGVERSION 9.4 + RUN apt-get install postgresql-${PGVERSION} -y + RUN apt-get install python python-psycopg2 python-yaml python-requests python-dnspython python-pip python-mock -y +-RUN pip install patroni ++## We install prereqs from pypi, the package from testpypi ++RUN pip install --force-reinstall --upgrade kazoo boto python-etcd ++RUN pip install -i https://testpypi.python.org/pypi patroni + + + ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index ce40f6dc..b28f4ae6 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -133,5 +133,5 @@ then sleep 60 done else - exec /patroni/patroni.py /patroni/postgres.yml + exec patroni /patroni/postgres.yml fi diff --git a/patroni/__init__.py b/patroni/__init__.py index a3e6290d..b970b513 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -1 +1 @@ -__version__ = '0.15' +__version__ = '0.22' diff --git a/patroni/helpers/__init__.py b/patroni/helpers/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/patroni/patroni.py b/patroni/patroni.py index dd7bd52e..2ee32211 100755 --- a/patroni/patroni.py +++ b/patroni/patroni.py @@ -5,12 +5,12 @@ import sys import time import yaml -from patroni.helpers.api import RestApiServer -from patroni.helpers.etcd import Etcd -from patroni.helpers.ha import Ha -from patroni.helpers.postgresql import Postgresql -from patroni.helpers.utils import setup_signal_handlers, sleep -from patroni.helpers.zookeeper import ZooKeeper +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.zookeeper import ZooKeeper logger = logging.getLogger(__name__) diff --git a/patroni/scripts/__init__.py b/patroni/scripts/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/requirements-py3.txt b/requirements-py3.txt index 544577ad..0fd9dfb3 100644 --- a/requirements-py3.txt +++ b/requirements-py3.txt @@ -5,5 +5,5 @@ psycopg2 PyYAML requests six -kazoo -python-etcd +kazoo>=2.2.1 +python-etcd>=0.4.1 diff --git a/setup.py b/setup.py index 82ac8de8..b90d1084 100644 --- a/setup.py +++ b/setup.py @@ -18,6 +18,7 @@ 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, '__init__.py'), 'r') as fd: @@ -56,6 +57,8 @@ CLASSIFIERS = [ 'Programming Language :: Python :: Implementation :: CPython', ] +CONSOLE_SCRIPTS = ['patroni = patroni.patroni:main'] + class PyTest(TestCommand): @@ -91,7 +94,6 @@ class PyTest(TestCommand): params['plugins'] = ['cov'] if self.junitxml: params['args'] += self.junitxml - #params['args'] += ['--doctest-modules', MAIN_PACKAGE, '--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s'] params['args'] += ['--doctest-modules', MAIN_PACKAGE, '-s', '-vv'] errno = pytest.main(**params) sys.exit(errno) @@ -144,6 +146,7 @@ def setup_package(): cmdclass=cmdclass, tests_require=['pytest-cov', 'pytest'], command_options=command_options, + entry_points={'console_scripts': CONSOLE_SCRIPTS}, ) From 04d6f7b418eb9fcecf87364ef23359b9abbd222a Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Thu, 3 Sep 2015 10:34:37 +0200 Subject: [PATCH 12/26] Make scripts into a module, so it can be imported. --- patroni/scripts/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 patroni/scripts/__init__.py diff --git a/patroni/scripts/__init__.py b/patroni/scripts/__init__.py new file mode 100644 index 00000000..e69de29b From 8984d991166ecf0b47c16b2ff6b39e4ff928f62f Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Thu, 3 Sep 2015 10:38:21 +0200 Subject: [PATCH 13/26] Make helpers into a module, so it can be imported. --- patroni/helpers/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 patroni/helpers/__init__.py diff --git a/patroni/helpers/__init__.py b/patroni/helpers/__init__.py new file mode 100644 index 00000000..e69de29b From 147d7c8566ba9b8ff2669b421d12e51cc3d63564 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Thu, 3 Sep 2015 10:45:06 +0200 Subject: [PATCH 14/26] Fix heading of the README --- README.rst | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/README.rst b/README.rst index c8c5eb7c..cb06c176 100644 --- a/README.rst +++ b/README.rst @@ -1,5 +1,8 @@ -|Build Status| |Coverage Status| # Patroni: A Template for PostgreSQL HA -with ZooKeeper or etcd +|Build Status| |Coverage Status| + +=== +Patroni: A Template for PostgreSQL HA with ZooKeeper or etcd +=== Patroni was previously known as Governor. From 58410db9dd804597b6c45cb5ec329f09eb362154 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 11:47:08 +0200 Subject: [PATCH 15/26] 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 66286733b2df8ed187e09aaa2a0259b86a4067cb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 15:03:09 +0200 Subject: [PATCH 16/26] 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 17/26] 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 18/26] 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 650e2449042763ccac6bd923f4549db9a5b31253 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 16:06:44 +0200 Subject: [PATCH 19/26] 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 5ea0ab70f558198869bf25f069e64f7c23a21eb8 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 7 Sep 2015 10:37:10 +0200 Subject: [PATCH 20/26] Add patroni.py executable file --- patroni.py | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100755 patroni.py diff --git a/patroni.py b/patroni.py new file mode 100755 index 00000000..f34ef133 --- /dev/null +++ b/patroni.py @@ -0,0 +1,6 @@ +#!/usr/bin/env python +from patroni import main + + +if __name__ == '__main__': + main() From 348e8e80864bb4d0acfe05f64e51fdf665a36dec Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 7 Sep 2015 13:47:05 +0200 Subject: [PATCH 21/26] Reverted pip installation in Dockerfile As the Dockerfile is there mainly to support developers, we want to build the Dockerfile using the current working directory instead of a previously released version. --- docker/Dockerfile => Dockerfile | 12 ++++++------ docker/Dockerfile.test.patch | 13 ------------- docker/entrypoint.sh | 2 +- patroni/version.py | 2 +- release.sh | 15 +++++++++------ 5 files changed, 17 insertions(+), 27 deletions(-) rename docker/Dockerfile => Dockerfile (86%) delete mode 100644 docker/Dockerfile.test.patch diff --git a/docker/Dockerfile b/Dockerfile similarity index 86% rename from docker/Dockerfile rename to Dockerfile index 1a1cb41c..61d85a74 100644 --- a/docker/Dockerfile +++ b/Dockerfile @@ -13,21 +13,21 @@ RUN apt-get update -y RUN apt-get upgrade -y ENV PGVERSION 9.4 -RUN apt-get install postgresql-${PGVERSION} -y -RUN apt-get install python python-psycopg2 python-yaml python-requests python-dnspython python-pip python-mock -y -RUN pip install patroni - +RUN apt-get install python python-psycopg2 python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo python-pip -y +RUN pip install python-etcd ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH +ADD patroni.py /patroni.py +ADD patroni/ /patroni + ENV ETCDVERSION 2.0.13 RUN curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz | tar xz -C /bin --strip=1 --wildcards --no-anchored etcd etcdctl ### Setting up a simple script that will serve as an entrypoint -RUN mkdir /patroni/ RUN mkdir /data/ && touch /var/log/etcd.log /var/log/etcd.err /pgpass /patroni/postgres.yml RUN chown postgres:postgres -R /patroni/ /data/ /pgpass /var/log/etcd.* /patroni/postgres.yml -ADD entrypoint.sh /entrypoint.sh +ADD docker/entrypoint.sh /entrypoint.sh EXPOSE 4001 5432 2380 diff --git a/docker/Dockerfile.test.patch b/docker/Dockerfile.test.patch deleted file mode 100644 index 376f3c20..00000000 --- a/docker/Dockerfile.test.patch +++ /dev/null @@ -1,13 +0,0 @@ ---- Dockerfile 2015-09-03 09:24:16.355412216 +0200 -+++ Dockerfile.test 2015-09-03 09:25:48.407879693 +0200 -@@ -15,7 +15,9 @@ - ENV PGVERSION 9.4 - RUN apt-get install postgresql-${PGVERSION} -y - RUN apt-get install python python-psycopg2 python-yaml python-requests python-dnspython python-pip python-mock -y --RUN pip install patroni -+## We install prereqs from pypi, the package from testpypi -+RUN pip install --force-reinstall --upgrade kazoo boto python-etcd -+RUN pip install -i https://testpypi.python.org/pypi patroni - - - ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index b28f4ae6..f4851a36 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -133,5 +133,5 @@ then sleep 60 done else - exec patroni /patroni/postgres.yml + exec python /patroni.py /patroni/postgres.yml fi diff --git a/patroni/version.py b/patroni/version.py index 11d27f8c..d0cc8295 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -1 +1 @@ -__version__ = '0.1' +__version__ = '0.45' diff --git a/release.sh b/release.sh index 316ca908..bf6f7482 100755 --- a/release.sh +++ b/release.sh @@ -5,6 +5,9 @@ if [ $# -ne 1 ]; then exit 1 fi +readonly VERSIONFILE="patroni/version.py" + +## Bail out on any non-zero exitcode from the called processes set -xe python3 --version @@ -12,17 +15,17 @@ git --version version=$1 -sed -i "s/__version__ = .*/__version__ = '${version}'/" version.py +sed -i "s/__version__ = .*/__version__ = '${version}'/" "${VERSIONFILE}" python3 setup.py clean python3 setup.py test python3 setup.py flake8 -git add __init__.py +git add "${VERSIONFILE}" -git commit -m "Bumped version to $version" -git push +#git commit -m "Bumped version to $version" +#git push python3 setup.py sdist bdist_wheel upload -git tag ${version} -git push --tags +#git tag ${version} +#git push --tags From cdccebd2d14c07ac50b574f834d82a6ce88432cc Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 7 Sep 2015 13:53:48 +0200 Subject: [PATCH 22/26] Point to api for listing of Docker images for Patroni (Issue #14) --- docker/README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docker/README.md b/docker/README.md index d17e77a1..74288a98 100644 --- a/docker/README.md +++ b/docker/README.md @@ -1,6 +1,7 @@ # Patroni Dockerfile -You can run Patroni in a docker container using this Dockerfile, or by using the Docker image at - https://os-registry.stups.zalan.do/acid/patroni-1.0-SNAPSHOT +You can run Patroni in a docker container using this Dockerfile, or by using one of the Docker image at + + https://os-registry.stups.zalan.do/v1/repositories/acid/patroni/tags This Dockerfile is meant in aiding development of Patroni and quick testing of features. It is not a production-worthy Dockerfile From 28f5839ac2fa4c8e730ec4e6a1c93ae0ce86775a Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 7 Sep 2015 14:00:05 +0200 Subject: [PATCH 23/26] Make git add part of the release.sh again. --- release.sh | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/release.sh b/release.sh index bf6f7482..e219603c 100755 --- a/release.sh +++ b/release.sh @@ -22,10 +22,10 @@ python3 setup.py flake8 git add "${VERSIONFILE}" -#git commit -m "Bumped version to $version" -#git push +git commit -m "Bumped version to $version" +git push python3 setup.py sdist bdist_wheel upload -#git tag ${version} -#git push --tags +git tag ${version} +git push --tags From 496f91fea077a450754dd049faf77c3d5287044f Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 8 Sep 2015 11:59:08 +0200 Subject: [PATCH 24/26] Update README.rst Formatting issue with the header --- README.rst | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/README.rst b/README.rst index 0798ad00..3e19f0a2 100644 --- a/README.rst +++ b/README.rst @@ -1,8 +1,7 @@ |Build Status| |Coverage Status| -=== Patroni: A Template for PostgreSQL HA with ZooKeeper or etcd -=== +------------------------------------------------------------ Patroni was previously known as Governor. From fa22d91e053f301498d9d09a950558758bf9b40f Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 8 Sep 2015 12:41:48 +0200 Subject: [PATCH 25/26] Move some basic methods implementation into parent exception class --- patroni/exceptions.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/patroni/exceptions.py b/patroni/exceptions.py index 5c13f3e0..e6159d47 100644 --- a/patroni/exceptions.py +++ b/patroni/exceptions.py @@ -1,8 +1,4 @@ class PatroniException(Exception): - pass - - -class DCSError(PatroniException): """Parent class for all kind of exceptions related to selected distributed configuration store""" @@ -15,3 +11,7 @@ class DCSError(PatroniException): "'foo'" """ return repr(self.value) + + +class DCSError(PatroniException): + pass From 61d1d5a098112e72aa1c1cfb657673b98428a12e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 8 Sep 2015 12:42:12 +0200 Subject: [PATCH 26/26] Set current version to 0.2 --- patroni/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/version.py b/patroni/version.py index d0cc8295..b650ceb0 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -1 +1 @@ -__version__ = '0.45' +__version__ = '0.2'