From b842ed478b6bf03cddf4da80385dedbd25104da7 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 8 Sep 2015 12:03:34 +0200 Subject: [PATCH 01/12] Make sure initialize flag is reset on failure. Cleanup the initialize flag if the initializing node fails to bootstrap its PostgreSQL database. Rename dcs.race to initialize, since we only call it for the initialize flag. Factored out PostgreSQL bootstrapping code into a separate function. --- helpers/dcs.py | 7 ++++++- helpers/etcd.py | 8 ++++++-- helpers/postgresql.py | 24 ++++++++++++++++++++++++ helpers/zookeeper.py | 9 +++++++-- patroni.py | 21 +++++++++++++-------- tests/test_etcd.py | 4 ++-- tests/test_patroni.py | 23 ++++++++++++++++++++--- tests/test_zookeeper.py | 2 +- 8 files changed, 79 insertions(+), 19 deletions(-) diff --git a/helpers/dcs.py b/helpers/dcs.py index c7140c22..fa812a08 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -66,6 +66,7 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem class AbstractDCS: __metaclass__ = abc.ABCMeta + initialize_key = '/initialize' def __init__(self, name, config): """ @@ -131,7 +132,7 @@ class AbstractDCS: overwriting the key if necessary.""" @abc.abstractmethod - def race(self, path): + def initialize(self): """Race for cluster initialization. :param path: usually this is just '/initialize' :returns: `!True` if key has been created successfully. @@ -144,5 +145,9 @@ class AbstractDCS: """Voluntarily remove leader key from DCS This method should remove leader key if current instance is the leader""" + @abc.abstractmethod + def cancel_initialization(self): + """ Removes the initialize key for a cluster """ + def sleep(self, timeout): sleep(timeout) diff --git a/helpers/etcd.py b/helpers/etcd.py index add736c5..93be5a05 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -206,9 +206,13 @@ class Etcd(AbstractDCS): return ret @catch_etcd_errors - def race(self, path): - return self.client.write(self.client_path(path), self._name, prevExist=False) + def initialize(self): + return self.client.write(self.client_path(self.initialize_key), self._name, prevExist=False) @catch_etcd_errors def delete_leader(self): return self.client.delete(self.client_path('/leader'), prevValue=self._name) + + @catch_etcd_errors + def cancel_initialization(self): + return self.client.delete(self.client_path(self.initialize_key), prevValue=self._name) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index e451d844..4e425dec 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -391,3 +391,27 @@ recovery_target_timeline = 'latest' def last_operation(self): return str(self.xlog_position()) + + def bootstrap(self, current_leader=None): + """ + Initially bootstrap PostgreSQL, either by creating a data + directory with initdb, or by initalizing a replica from an + exiting leader. Failure in the first case always leads to + exception, since there is no point in continuing if initdb failed. + In the second case, however, a False is returned on failure, since + it is normal for the replica to retry a failed attempt to initialize + from the master. + """ + ret = False + if not current_leader: + ret = self.initialize() and self.start() + if ret: + self.create_replication_user() + self.create_connection_users() + else: + raise Exception("Could not bootstrap master PostgreSQL") + else: + if self.sync_from_leader(current_leader): + self.write_recovery_conf(current_leader) + ret = self.start() + return ret diff --git a/helpers/zookeeper.py b/helpers/zookeeper.py index cb2918cd..98d68ad1 100644 --- a/helpers/zookeeper.py +++ b/helpers/zookeeper.py @@ -180,8 +180,8 @@ class ZooKeeper(AbstractDCS): ret or logger.info('Could not take out TTL lock') return ret - def race(self, path): - return self._create(path, self._name, makepath=True) + def initialize(self): + return self._create(self.initialize_key, self._name, makepath=True) def touch_member(self, connection_string, ttl=None): for m in self.members: @@ -223,6 +223,11 @@ class ZooKeeper(AbstractDCS): if isinstance(self.leader, Member) and self.leader.name == self._name: self.client.delete(self.client_path('/leader')) + def cancel_initialization(self): + node = self.get_node(self.initialize_key) + if node and node == self._name: + self.client.delete(self.client_path(self.initialize_key)) + def sleep(self, timeout): self.cluster_event.wait(timeout) if self.cluster_event.isSet(): diff --git a/patroni.py b/patroni.py index 3be2aa57..64430318 100755 --- a/patroni.py +++ b/patroni.py @@ -43,6 +43,11 @@ class Patroni: return True return self.ha.dcs.touch_member(connection_string, ttl) + def cleanup_on_failed_initialization(self): + """ cleanup the DCS if initialization was not successfull """ + logger.info("removing initialize key after failed attempt to initialize the cluster") + self.ha.dcs.cancel_initialization() + def initialize(self): # wait for etcd to be available while not self.touch_member(): @@ -52,18 +57,18 @@ class Patroni: # is data directory empty? if self.postgresql.data_directory_empty(): # racing to initialize - if self.ha.dcs.race('/initialize'): - self.postgresql.initialize() + if self.ha.dcs.initialize(): + try: + self.postgresql.bootstrap() + except: + # bail out and clean the initialize flag. + self.cleanup_on_failed_initialization() + raise 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() + if leader and self.postgresql.bootstrap(leader): break sleep(5) elif self.postgresql.is_running(): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 38692c52..fa534d15 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -227,8 +227,8 @@ class TestEtcd(unittest.TestCase): def test_update_leader(self): self.assertTrue(self.etcd.update_leader(MockPostgresql())) - def test_race(self): - self.assertFalse(self.etcd.race('')) + def test_initialize(self): + self.assertFalse(self.etcd.initialize()) def test_delete_leader(self): self.etcd.client.delete = etcd_delete diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 67ac1ffd..eb41591c 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -46,6 +46,7 @@ class TestPatroni(unittest.TestCase): def set_up(self): self.touched = False + self.init_cancelled = False subprocess.call = subprocess_call psycopg2.connect = psycopg2_connect self.time_sleep = time.sleep @@ -121,14 +122,15 @@ class TestPatroni(unittest.TestCase): self.p.touch_member() def test_patroni_initialize(self): - self.p.postgresql.should_use_s3_to_create_replica = false self.p.ha.dcs.client.write = etcd_write self.p.touch_member = self.touch_member self.p.postgresql.data_directory_empty = true - self.p.ha.dcs.race = true + self.p.ha.dcs.initialize = true + self.p.postgresql.initialize = true + self.p.postgresql.start = true self.p.initialize() - self.p.ha.dcs.race = false + self.p.ha.dcs.initialize = false time.sleep = time_sleep self.p.ha.dcs.client.read = etcd_read self.p.initialize() @@ -142,3 +144,18 @@ class TestPatroni(unittest.TestCase): def test_schedule_next_run(self): self.p.next_run = time.time() - self.p.nap_time - 1 self.p.schedule_next_run() + + def cancel_initialization(self): + self.init_cancelled = True + + def test_cleanup_on_initialization(self): + self.p.ha.dcs.client.write = etcd_write + self.p.touch_member = self.touch_member + self.p.postgresql.data_directory_empty = true + self.p.ha.dcs.initialize = true + self.p.postgresql.initialize = true + self.p.postgresql.start = false + + self.p.ha.dcs.cancel_initialization = self.cancel_initialization + self.assertRaises(Exception, self.p.initialize) + self.assertTrue(self.init_cancelled) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 53f5ab82..28b337a9 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -141,7 +141,7 @@ class TestZooKeeper(unittest.TestCase): self.zk.delete_leader() def test_race(self): - self.assertFalse(self.zk.race('/initialize')) + self.assertFalse(self.zk.initialize()) def test_touch_member(self): self.zk.touch_member('new') From ff499604f0cfc5ce1b9e581ce3838ef943d78860 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 8 Sep 2015 16:04:54 +0200 Subject: [PATCH 02/12] Act on removal of initialization flag. If initializer node suddenly dies before the initialization is complete, other nodes should try to take over. Fix some unittests for etcd and zookeeper and add couple of new ones. --- patroni/__init__.py | 23 ++++++++++++----------- tests/test_etcd.py | 4 ++++ tests/test_zookeeper.py | 9 ++++++++- 3 files changed, 24 insertions(+), 12 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 46c030b6..b55c0dfc 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -55,17 +55,18 @@ class Patroni: # is data directory empty? if self.postgresql.data_directory_empty(): - # racing to initialize - if self.ha.dcs.initialize(): - try: - self.postgresql.bootstrap() - except: - # bail out and clean the initialize flag. - self.cleanup_on_failed_initialization() - raise - self.ha.dcs.take_leader() - else: - while True: + while True: + # racing to initialize + if self.ha.dcs.initialize(): + try: + self.postgresql.bootstrap() + except: + # bail out and clean the initialize flag. + self.cleanup_on_failed_initialization() + raise + self.ha.dcs.take_leader() + break + else: leader = self.ha.dcs.current_leader() if leader and self.postgresql.bootstrap(leader): break diff --git a/tests/test_etcd.py b/tests/test_etcd.py index b5c014f4..dbcd4b08 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -269,6 +269,10 @@ class TestEtcd(unittest.TestCase): def test_initialize(self): self.assertFalse(self.etcd.initialize()) + def test_cancel_initializion(self): + self.etcd.client.delete = etcd_delete + self.assertFalse(self.etcd.cancel_initialization()) + def test_delete_leader(self): self.etcd.client.delete = etcd_delete self.assertFalse(self.etcd.delete_leader()) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index d138e4b2..83c97b7f 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -71,6 +71,8 @@ class MockKazooClient: if self.leader: return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0)) return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) + elif path.endswith(patroni.zookeeper.ZooKeeper.initialize_key): + return 'foo' def get_children(self, path, watch=None, include_data=False): return ['foo', 'bar', 'buzz'] @@ -93,6 +95,8 @@ class MockKazooClient: return self.leader = True raise Exception + elif path.endswith(patroni.zookeeper.ZooKeeper.initialize_key): + raise Exception def set_hosts(self, hosts, randomize_hosts=None): pass @@ -146,9 +150,12 @@ class TestZooKeeper(unittest.TestCase): self.zk.touch_member('foo') self.zk.delete_leader() - def test_race(self): + def test_initialize(self): self.assertFalse(self.zk.initialize()) + def test_cancel_initialization(self): + self.assertRaises(Exception, self.zk.cancel_initialization) + def test_touch_member(self): self.zk.touch_member('new') self.zk.touch_member('exists') From 30a9e0f7f5da186fdbc855f8f26d18bc6fde80b9 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 10 Sep 2015 15:34:29 +0200 Subject: [PATCH 03/12] Move PostgreSQL data directory if init had failed. Prevent treating the incompletely-initialized PostgreSQL cluster as a valid on restart by forcefully moving the data directory. I don't want to remove it altogether, since a DBA might decide to analyze the failed PG cluster in order to resolve the init issue. --- patroni/__init__.py | 2 ++ patroni/postgresql.py | 9 ++++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index b55c0dfc..6a4bf6aa 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -46,6 +46,8 @@ class Patroni: """ cleanup the DCS if initialization was not successfull """ logger.info("removing initialize key after failed attempt to initialize the cluster") self.ha.dcs.cancel_initialization() + self.postgresql.stop() + self.postgresql.move_data_directory() def initialize(self): # wait for etcd to be available diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 8754bb65..3c1630a8 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -4,6 +4,7 @@ import psycopg2 import shlex import shutil import subprocess +import time from patroni.utils import sleep from six.moves.urllib_parse import urlparse @@ -159,7 +160,7 @@ class Postgresql: return ret def is_running(self): - return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0 + return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null 2>&1', shell=True) == 0 def call_nowait(self, cb_name, is_leader=None): """ pick a callback command and call it without waiting for it to finish """ @@ -415,3 +416,9 @@ recovery_target_timeline = 'latest' self.write_recovery_conf(current_leader) ret = self.start() return ret + + def move_data_directory(self): + if os.path.isdir(self.data_dir) and not self.is_running(): + os.rename(self.data_dir, '{0}_{1}'.format(self.data_dir, str(long(time.time())))) + return True + return False From 2377c417e40bbbcc64173a64ab989dc8c894c10b Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 10 Sep 2015 16:05:10 +0200 Subject: [PATCH 04/12] Fix etcd and zookeper interactions with initialize key. Fix unittests as well. --- patroni/dcs.py | 1 - patroni/etcd.py | 4 ++-- patroni/zookeeper.py | 9 ++++++--- tests/test_zookeeper.py | 10 +++++----- 4 files changed, 13 insertions(+), 11 deletions(-) diff --git a/patroni/dcs.py b/patroni/dcs.py index 0c2bb9f7..bdba4004 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -73,7 +73,6 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem class AbstractDCS: __metaclass__ = abc.ABCMeta - initialize_key = '/initialize' _INITIALIZE = 'initialize' _LEADER = 'leader' diff --git a/patroni/etcd.py b/patroni/etcd.py index d5730ffe..e24fa66b 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -233,7 +233,7 @@ class Etcd(AbstractDCS): @catch_etcd_errors def initialize(self): - return self.client.write(self.client_path(self.initialize_key), self._name, prevExist=False) + return self.client.write(self.initialize_path, self._name, prevExist=False) @catch_etcd_errors def delete_leader(self): @@ -241,7 +241,7 @@ class Etcd(AbstractDCS): @catch_etcd_errors def cancel_initialization(self): - return self.client.delete(self.client_path(self.initialize_key), prevValue=self._name) + return self.client.delete(self.initialize_path, prevValue=self._name) def watch(self, timeout): # watch on leader key changes if it is defined and current node is not lock owner diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 7245b1a3..7c01f278 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -4,7 +4,7 @@ import requests import time from kazoo.client import KazooClient, KazooState -from kazoo.exceptions import NoNodeError, NodeExistsError +from kazoo.exceptions import NoNodeError, NodeExistsError, KazooException from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string from patroni.utils import sleep from requests.exceptions import RequestException @@ -222,8 +222,11 @@ class ZooKeeper(AbstractDCS): def cancel_initialization(self): node = self.get_node(self.initialize_path) - if node and node == self._name: - self.client.delete(self.initialize_path) + if node and node[0] == self._name: + try: + self.client.retry(self.client.delete, self.initialize_path, version=node[1].mzxid) + except KazooException: + logger.exception("Unable to delete initialize key") def watch(self, timeout): self.cluster_event.wait(timeout) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 3da71f02..0a95d0bc 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -71,8 +71,8 @@ class MockKazooClient: if self.leader: return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0)) return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) - elif path.endswith(patroni.zookeeper.ZooKeeper.initialize_key): - return 'foo' + elif path.endswith('/initialize'): + return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) def get_children(self, path, watch=None, include_data=False): return ['foo', 'bar', 'buzz'] @@ -95,8 +95,8 @@ class MockKazooClient: return self.leader = True raise Exception - elif path.endswith(patroni.zookeeper.ZooKeeper.initialize_key): - raise Exception + elif path.endswith('/initialize'): + raise NoNodeError def set_hosts(self, hosts, randomize_hosts=None): pass @@ -154,7 +154,7 @@ class TestZooKeeper(unittest.TestCase): self.assertFalse(self.zk.initialize()) def test_cancel_initialization(self): - self.assertRaises(Exception, self.zk.cancel_initialization) + self.zk.cancel_initialization() def test_touch_member(self): self.zk.touch_member('new') From cd312de2527e0d168ff0fad27978120c17919383 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 10 Sep 2015 17:15:43 +0200 Subject: [PATCH 05/12] Fix a flake8 warning. Improve some unit tests by expecting specific exceptions. --- patroni/exceptions.py | 4 ++++ patroni/postgresql.py | 10 ++++++---- tests/test_patroni.py | 5 +++-- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/patroni/exceptions.py b/patroni/exceptions.py index e6159d47..507edfc3 100644 --- a/patroni/exceptions.py +++ b/patroni/exceptions.py @@ -13,5 +13,9 @@ class PatroniException(Exception): return repr(self.value) +class PostgresException(PatroniException): + pass + + class DCSError(PatroniException): pass diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 3c1630a8..b7f76fc3 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -6,6 +6,7 @@ import shutil import subprocess import time +from patroni.exceptions import PostgresException from patroni.utils import sleep from six.moves.urllib_parse import urlparse @@ -410,7 +411,7 @@ recovery_target_timeline = 'latest' self.create_replication_user() self.create_connection_users() else: - raise Exception("Could not bootstrap master PostgreSQL") + raise PostgresException("Could not bootstrap master PostgreSQL") else: if self.sync_from_leader(current_leader): self.write_recovery_conf(current_leader) @@ -419,6 +420,7 @@ recovery_target_timeline = 'latest' def move_data_directory(self): if os.path.isdir(self.data_dir) and not self.is_running(): - os.rename(self.data_dir, '{0}_{1}'.format(self.data_dir, str(long(time.time())))) - return True - return False + try: + os.rename(self.data_dir, '{0}_{1}'.format(self.data_dir, time.strftime('%Y-%m-%d-%H-%M-%S'))) + except: + logger.exception("Could not rename data directory {0}".format(self.data_dir)) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 75d47fe4..d5740e9c 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -11,6 +11,7 @@ from mock import Mock, patch from patroni.api import RestApiServer from patroni.dcs import Cluster, Member from patroni.etcd import Etcd +from patroni.exceptions import PostgresException from patroni import Patroni, main from patroni.zookeeper import ZooKeeper from six.moves import BaseHTTPServer @@ -141,7 +142,7 @@ class TestPatroni(unittest.TestCase): self.p.initialize() self.p.ha.dcs.current_leader = nop - self.assertRaises(Exception, self.p.initialize) + self.assertRaises(SleepException, self.p.initialize) self.p.postgresql.data_directory_empty = false self.p.initialize() @@ -162,5 +163,5 @@ class TestPatroni(unittest.TestCase): self.p.postgresql.start = false self.p.ha.dcs.cancel_initialization = self.cancel_initialization - self.assertRaises(Exception, self.p.initialize) + self.assertRaises(PostgresException, self.p.initialize) self.assertTrue(self.init_cancelled) From 15cd10669dc632a78c8ff7d27c6c3c25f452df32 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 10 Sep 2015 18:04:47 +0200 Subject: [PATCH 06/12] Change an outdated comment. --- patroni/dcs.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/patroni/dcs.py b/patroni/dcs.py index bdba4004..9017da7f 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -167,10 +167,9 @@ class AbstractDCS: @abc.abstractmethod def initialize(self): """Race for cluster initialization. - :param path: usually this is just '/initialize' :returns: `!True` if key has been created successfully. - this method should create atomically `path` key and return `!True` + this method should create atomically initialize key and return `!True` otherwise it should return `!False`""" @abc.abstractmethod From be110c4ba0036056297de02e3fc38d7c2975bfbd Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 14 Sep 2015 09:20:45 +0200 Subject: [PATCH 07/12] Do not try to stop postgres twice if initialization had failed. --- patroni/__init__.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 6a4bf6aa..6406143b 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -46,6 +46,7 @@ class Patroni: """ cleanup the DCS if initialization was not successfull """ logger.info("removing initialize key after failed attempt to initialize the cluster") self.ha.dcs.cancel_initialization() + self.touch_member(self.shutdown_member_ttl) self.postgresql.stop() self.postgresql.move_data_directory() @@ -118,8 +119,8 @@ def main(): config = yaml.load(f) patroni = Patroni(config) + patroni.initialize() try: - patroni.initialize() patroni.run() except KeyboardInterrupt: pass From f494d2ce646dfdc35771eb8472dfb543e264829e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 14 Sep 2015 11:19:46 +0200 Subject: [PATCH 08/12] Build Cluster object for ZooKeeper the same way as for Etcd Previous implementation was always setting Cluster.initialize to True. Also it was throwing ZooKeeperError when there were no members in a cluster. Plus BUGFIX of a bug introduced with https://github.com/zalando/patroni/pull/34 in a `load_members` method. - data = self.get_node(self.member_path) + data = self.get_node(self.members_path + member) It was always fetching the same node for all cluster members. Fortunately Etcd doesn't have such problem because we are fetching the whole cluster directory with one recursive API call. --- patroni/zookeeper.py | 50 ++++++++++++++++++++++++++--------------- tests/test_zookeeper.py | 10 +++++++++ 2 files changed, 42 insertions(+), 18 deletions(-) diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 7c01f278..5f9625ac 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -92,9 +92,8 @@ class ZooKeeper(AbstractDCS): self.client.add_listener(self.session_listener) self.cluster_event = self.client.handler.event_object() + self.cluster = None self.fetch_cluster = True - self.members = [] - self.leader = None self.last_leader_operation = 0 self.client.start(None) @@ -121,18 +120,35 @@ class ZooKeeper(AbstractDCS): conn_url, api_url = parse_connection_string(value) return Member(znode.mzxid, name, conn_url, api_url, None, None) + def get_children(self, key, watch=None): + try: + return self.client.get_children(key, watch) + except NoNodeError: + pass + except: + logger.exception('get_children') + return [] + def load_members(self): members = [] - for member in self.client.get_children(self.members_path, self.cluster_watcher): - data = self.get_node(self.member_path) + for member in self.get_children(self.members_path, self.cluster_watcher): + data = self.get_node(self.members_path + member) if data is not None: members.append(self.member(member, *data)) return members def _inner_load_cluster(self): self.cluster_event.clear() - leader = self.get_node(self.leader_path, self.cluster_watcher) - self.members = self.load_members() + nodes = set(self.get_children(self.client_path(''))) + + # get initialize flag + initialize = self._INITIALIZE in nodes + + # get list of members + members = self.load_members() if self._MEMBERS[:-1] in nodes else [] + + # get leader + leader = self.get_node(self.leader_path, self.cluster_watcher) if self._LEADER in nodes else None if leader: client_id = self.client.client_id if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner: @@ -142,15 +158,14 @@ class ZooKeeper(AbstractDCS): if leader: member = Member(-1, leader[0], None, None, None, None) - member = ([m for m in self.members if m.name == leader[0]] or [member])[0] + member = ([m for m in 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(self.leader_optime_path) - if last_leader_operation: - self.last_leader_operation = int(last_leader_operation[0]) + # get last leader operation + self.last_leader_operation = self.get_node(self.leader_optime_path) if self.fetch_cluster else None + self.last_leader_operation = 0 if self.last_leader_operation is None else int(self.last_leader_operation[0]) + self.cluster = Cluster(initialize, leader, self.last_leader_operation, members) def get_cluster(self): if self.exhibitor and self.exhibitor.poll(): @@ -163,7 +178,7 @@ class ZooKeeper(AbstractDCS): logger.exception('get_cluster') self.session_listener(KazooState.LOST) raise ZooKeeperError('ZooKeeper in not responding properly') - return Cluster(True, self.leader, self.last_leader_operation, self.members) + return self.cluster def _create(self, path, value, **kwargs): try: @@ -181,9 +196,8 @@ class ZooKeeper(AbstractDCS): return self._create(self.initialize_path, self._name, makepath=True) def touch_member(self, connection_string, ttl=None): - for m in self.members: - if m.name == self._name: - return True + if self.cluster and any(m.name == self._name for m in self.cluster.members): + return True path = self.member_path try: self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True) @@ -217,8 +231,8 @@ class ZooKeeper(AbstractDCS): return True def delete_leader(self): - if isinstance(self.leader, Leader) and self.leader.name == self._name: - self.client.delete(self.leader_path) + if isinstance(self.cluster, Cluster) and self.cluster.leader.name == self._name: + self.client.delete(self.leader_path, version=self.cluster.leader.index) def cancel_initialization(self): node = self.get_node(self.initialize_path) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 0a95d0bc..19435c6e 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -75,6 +75,12 @@ class MockKazooClient: return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) def get_children(self, path, watch=None, include_data=False): + if path == '/no_node': + raise NoNodeError + elif path == '/other_exception': + raise Exception() + elif path in ['/service/bla/', '/service/test/']: + return ['initialize', 'leader', 'members', 'optime'] return ['foo', 'bar', 'buzz'] def create(self, path, value="", acl=None, ephemeral=False, sequence=False, makepath=False): @@ -138,6 +144,10 @@ class TestZooKeeper(unittest.TestCase): self.assertIsNone(self.zk.get_node('/no_node')) self.assertIsNone(self.zk.get_node('/other_exception')) + def test_get_children(self): + self.assertListEqual(self.zk.get_children('/no_node'), []) + self.assertListEqual(self.zk.get_children('/other_exception'), []) + def test__inner_load_cluster(self): self.zk._base_path = self.zk._base_path.replace('test', 'bla') self.zk._inner_load_cluster() From 209c985420b5b437b8afc9dfedfabf7366a01944 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 14 Sep 2015 11:45:00 +0200 Subject: [PATCH 09/12] get_node and get_children should catch only NoNodeError exception. All other exceptions are needed to have retry functionality working correctly. --- patroni/zookeeper.py | 10 ++-------- tests/test_zookeeper.py | 6 ------ 2 files changed, 2 insertions(+), 14 deletions(-) diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 5f9625ac..f2ef2e99 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -110,10 +110,7 @@ class ZooKeeper(AbstractDCS): try: return self.client.get(key, watch) except NoNodeError: - pass - except: - logger.exception('get_node') - return None + return None @staticmethod def member(name, value, znode): @@ -124,10 +121,7 @@ class ZooKeeper(AbstractDCS): try: return self.client.get_children(key, watch) except NoNodeError: - pass - except: - logger.exception('get_children') - return [] + return [] def load_members(self): members = [] diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 19435c6e..4b31bad2 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -58,8 +58,6 @@ class MockKazooClient: def get(self, path, watch=None): if path == '/no_node': raise NoNodeError - elif path == '/other_exception': - raise Exception() elif '/members/' in path: return ( 'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni', @@ -77,8 +75,6 @@ class MockKazooClient: def get_children(self, path, watch=None, include_data=False): if path == '/no_node': raise NoNodeError - elif path == '/other_exception': - raise Exception() elif path in ['/service/bla/', '/service/test/']: return ['initialize', 'leader', 'members', 'optime'] return ['foo', 'bar', 'buzz'] @@ -142,11 +138,9 @@ class TestZooKeeper(unittest.TestCase): def test_get_node(self): self.assertIsNone(self.zk.get_node('/no_node')) - self.assertIsNone(self.zk.get_node('/other_exception')) def test_get_children(self): self.assertListEqual(self.zk.get_children('/no_node'), []) - self.assertListEqual(self.zk.get_children('/other_exception'), []) def test__inner_load_cluster(self): self.zk._base_path = self.zk._base_path.replace('test', 'bla') From 4a081bcb7179ad00f3771113176c27706ca43a76 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 14 Sep 2015 11:58:10 +0200 Subject: [PATCH 10/12] Run cancel_initialization with retry --- patroni/zookeeper.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index f2ef2e99..38891c58 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -228,13 +228,16 @@ class ZooKeeper(AbstractDCS): if isinstance(self.cluster, Cluster) and self.cluster.leader.name == self._name: self.client.delete(self.leader_path, version=self.cluster.leader.index) - def cancel_initialization(self): + def _cancel_initialization(self): node = self.get_node(self.initialize_path) if node and node[0] == self._name: - try: - self.client.retry(self.client.delete, self.initialize_path, version=node[1].mzxid) - except KazooException: - logger.exception("Unable to delete initialize key") + self.client.delete(self.initialize_path, version=node[1].mzxid) + + def cancel_initialization(self): + try: + self.client.retry(self._cancel_initialization) + except: + logger.exception("Unable to delete initialize key") def watch(self, timeout): self.cluster_event.wait(timeout) From 98488a00a232d0294a2c4f5edc6414bf5a271d2b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 14 Sep 2015 12:00:24 +0200 Subject: [PATCH 11/12] Remove unused import of KazooException --- patroni/zookeeper.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 38891c58..708cbafa 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -4,7 +4,7 @@ import requests import time from kazoo.client import KazooClient, KazooState -from kazoo.exceptions import NoNodeError, NodeExistsError, KazooException +from kazoo.exceptions import NoNodeError, NodeExistsError from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string from patroni.utils import sleep from requests.exceptions import RequestException From 51eacc5042bb4548a9bd5339497631f4c46ba90a Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 14 Sep 2015 12:36:28 +0200 Subject: [PATCH 12/12] Handle the case when initialize flag is not set and leader is present. --- patroni/__init__.py | 34 ++++++++++++++++++++-------------- tests/test_patroni.py | 37 +++++++++++++++++++++++++++++++++++-- tests/test_postgresql.py | 7 +++++++ 3 files changed, 62 insertions(+), 16 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 6406143b..cc008fda 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -6,6 +6,7 @@ import yaml from patroni.api import RestApiServer from patroni.etcd import Etcd +from patroni.exceptions import DCSError from patroni.ha import Ha from patroni.postgresql import Postgresql from patroni.utils import setup_signal_handlers, sleep, reap_children @@ -59,21 +60,26 @@ class Patroni: # is data directory empty? if self.postgresql.data_directory_empty(): while True: - # racing to initialize - if self.ha.dcs.initialize(): - try: - self.postgresql.bootstrap() - except: - # bail out and clean the initialize flag. - self.cleanup_on_failed_initialization() - raise - self.ha.dcs.take_leader() - break - else: - leader = self.ha.dcs.current_leader() - if leader and self.postgresql.bootstrap(leader): + try: + cluster = self.ha.dcs.get_cluster() + if not cluster.is_unlocked(): # the leader already exists + if not cluster.initialize: + self.ha.dcs.initialize() + self.postgresql.bootstrap(cluster.leader) break - sleep(5) + # racing to initialize + elif not cluster.initialize and self.ha.dcs.initialize(): + try: + self.postgresql.bootstrap() + except: + # bail out and clean the initialize flag. + self.cleanup_on_failed_initialization() + raise + self.ha.dcs.take_leader() + break + except DCSError: + logger.info('waiting on DCS') + sleep(5) elif self.postgresql.is_running(): self.postgresql.load_replication_slots() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index d5740e9c..317e1e5c 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -9,7 +9,7 @@ import yaml from mock import Mock, patch from patroni.api import RestApiServer -from patroni.dcs import Cluster, Member +from patroni.dcs import Cluster, Member, Leader from patroni.etcd import Etcd from patroni.exceptions import PostgresException from patroni import Patroni, main @@ -42,6 +42,30 @@ class Mock_BaseServer__is_shut_down: pass +def get_cluster(initialize, leader): + return Cluster(initialize, leader, None, None) + + +def get_cluster_not_initialized_without_leader(): + return get_cluster(None, None) + + +def get_cluster_initialized_without_leader(): + return get_cluster(True, None) + + +def get_cluster_not_initialized_with_leader(): + return get_cluster(False, Leader(0, 0, 0, + Member(0, 'leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', + None, None, 28))) + + +def get_cluster_initialized_with_leader(): + return get_cluster(True, Leader(0, 0, 0, + Member(0, 'leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', + None, None, 28))) + + class TestPatroni(unittest.TestCase): def __init__(self, method_name='runTest'): @@ -129,24 +153,31 @@ class TestPatroni(unittest.TestCase): def test_patroni_initialize(self): self.p.ha.dcs.client.write = etcd_write + self.p.ha.dcs.client.read = etcd_read self.p.touch_member = self.touch_member self.p.postgresql.data_directory_empty = true self.p.ha.dcs.initialize = true self.p.postgresql.initialize = true self.p.postgresql.start = true + self.p.ha.dcs.get_cluster = get_cluster_not_initialized_without_leader self.p.initialize() self.p.ha.dcs.initialize = false + self.p.ha.dcs.get_cluster = get_cluster_initialized_with_leader time.sleep = time_sleep self.p.ha.dcs.client.read = etcd_read self.p.initialize() - self.p.ha.dcs.current_leader = nop + self.p.ha.dcs.get_cluster = get_cluster_initialized_without_leader self.assertRaises(SleepException, self.p.initialize) self.p.postgresql.data_directory_empty = false self.p.initialize() + self.p.ha.dcs.get_cluster = get_cluster_not_initialized_with_leader + self.p.postgresql.data_directory_empty = true + self.p.initialize() + def test_schedule_next_run(self): self.p.next_run = time.time() - self.p.nap_time - 1 self.p.schedule_next_run() @@ -156,6 +187,8 @@ class TestPatroni(unittest.TestCase): def test_cleanup_on_initialization(self): self.p.ha.dcs.client.write = etcd_write + self.p.ha.dcs.client.read = etcd_read + self.p.ha.dcs.get_cluster = get_cluster_not_initialized_without_leader self.p.touch_member = self.touch_member self.p.postgresql.data_directory_empty = true self.p.ha.dcs.initialize = true diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 273d077d..56dbc557 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -6,6 +6,7 @@ import unittest from patroni.dcs import Cluster, Leader, Member from patroni.postgresql import Postgresql +from test_ha import true, false def nop(*args, **kwargs): @@ -217,3 +218,9 @@ class TestPostgresql(unittest.TestCase): self.p.start() self.p.query = self.mock_query self.assertTrue(self.p.stop()) + + def test_move_data_directory(self): + self.p.is_running = is_running + os.rename = nop + os.path.isdir = true + self.p.move_data_directory()